title: OpenMessaging兼容示例 date: 2017/12/21

categories: 文档翻译

OpenMessaging Example

OpenMessaging, which includes the establishment of industry guidelines and messaging, streaming specifications to provide a common framework for finance, e-commerce, IoT and big-data area. The design principles are the cloud-oriented, simplicity, flexibility, and language independent in distributed heterogeneous environments. Conformance to these specifications will make it possible to develop a heterogeneous messaging applications across all major platforms and operating systems.

OpenMessaging兼容示例

OpenMessaging力图建立行业准则和消息分发、流计算领域标准，为金融、电子商务、物联网和大数据领域提供通用框架。设计原则包括，面向云、简单易用、灵活度高、独立于编程语言，以及能在分布式异构环境中使用。这些标准的一致性将会使得开发一款跨所有主流平台和操作系统的异构消息分发应用成为可能。

RocketMQ provides a partial implementation of OpenMessaging 0.1.0-alpha, the following examples demonstrate how to access RocketMQ based on OpenMessaging.

RocketMQ提供了关于OpenMessaging 0.1.0-alpha版本的部分实现，以下例子展示了如何按照OpenMessaging规范接入RocketMQ.

OMSProducer

The following example shows how to send message to RocketMQ broker in synchronous, asynchronous, or one-way transmissions.

  1. public class OMSProducer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final Producer producer = messagingAccessPoint.createProducer();
  8.         messagingAccessPoint.startup();
  9.         System.out.printf("MessagingAccessPoint startup OK%n");
  11.         producer.startup();
  12.         System.out.printf("Producer startup OK%n");
  14.         {
  15.             Message message = producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
  16.             SendResult sendResult = producer.send(message);
  17.             System.out.printf("Send sync message OK, msgId: %s%n", sendResult.messageId());
  18.         }
  20.         {
  21.             final Promise<SendResult> result = producer.sendAsync(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
  22.             result.addListener(new PromiseListener<SendResult>() {
  23.                 @Override
  24.                 public void operationCompleted(Promise<SendResult> promise) {
  25.                     System.out.printf("Send async message OK, msgId: %s%n", promise.get().messageId());
  26.                 }
  28.                 @Override
  29.                 public void operationFailed(Promise<SendResult> promise) {
  30.                     System.out.printf("Send async message Failed, error: %s%n", promise.getThrowable().getMessage());
  31.                 }
  32.             });
  33.         }
  35.         {
  36.             producer.sendOneway(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
  37.             System.out.printf("Send oneway message OK%n");
  38.         }
  40.         producer.shutdown();
  41.         messagingAccessPoint.shutdown();
  42.     }
  43. }

OMSProducer

下面这个例子展示了如何向RocketMQ broker用同步、异步、单向发送的方式发送消息。

  1. public class OMSProducer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final Producer producer = messagingAccessPoint.createProducer();
  8.         messagingAccessPoint.startup();
  9.         System.out.printf("MessagingAccessPoint startup OK%n");
  11.         producer.startup();
  12.         System.out.printf("Producer startup OK%n");
  14.         {
  15.             Message message = producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
  16.             SendResult sendResult = producer.send(message);
  17.             System.out.printf("Send sync message OK, msgId: %s%n", sendResult.messageId());
  18.         }
  20.         {
  21.             final Promise<SendResult> result = producer.sendAsync(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
  22.             result.addListener(new PromiseListener<SendResult>() {
  23.                 @Override
  24.                 public void operationCompleted(Promise<SendResult> promise) {
  25.                     System.out.printf("Send async message OK, msgId: %s%n", promise.get().messageId());
  26.                 }
  28.                 @Override
  29.                 public void operationFailed(Promise<SendResult> promise) {
  30.                     System.out.printf("Send async message Failed, error: %s%n", promise.getThrowable().getMessage());
  31.                 }
  32.             });
  33.         }
  35.         {
  36.             producer.sendOneway(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
  37.             System.out.printf("Send oneway message OK%n");
  38.         }
  40.         producer.shutdown();
  41.         messagingAccessPoint.shutdown();
  42.     }
  43. }

OMSPullConsumer

Use OMS PullConsumer to poll【译者注：为什么是poll】 messages from a specified queue.

  1. public class OMSPullConsumer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final PullConsumer consumer = messagingAccessPoint.createPullConsumer("OMS_HELLO_TOPIC",
  7.             OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
  9.         messagingAccessPoint.startup();
  10.         System.out.printf("MessagingAccessPoint startup OK%n");
  12.         consumer.startup();
  13.         System.out.printf("Consumer startup OK%n");
  15.         Message message = consumer.poll();
  16.         if (message != null) {
  17.             String msgId = message.headers().getString(MessageHeader.MESSAGE_ID);
  18.             System.out.printf("Received one message: %s%n", msgId);
  19.             consumer.ack(msgId);
  20.         }
  22.         consumer.shutdown();
  23.         messagingAccessPoint.shutdown();
  24.     }
  25. }

OMSPullConsumer

使用OMS PullConsumer从特定队列中拉取消息。

  1. public class OMSPullConsumer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final PullConsumer consumer = messagingAccessPoint.createPullConsumer("OMS_HELLO_TOPIC",
  7.             OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
  9.         messagingAccessPoint.startup();
  10.         System.out.printf("MessagingAccessPoint startup OK%n");
  12.         consumer.startup();
  13.         System.out.printf("Consumer startup OK%n");
  15.         Message message = consumer.poll();
  16.         if (message != null) {
  17.             String msgId = message.headers().getString(MessageHeader.MESSAGE_ID);
  18.             System.out.printf("Received one message: %s%n", msgId);
  19.             consumer.ack(msgId);
  20.         }
  22.         consumer.shutdown();
  23.         messagingAccessPoint.shutdown();
  24.     }
  25. }

OMSPushConsumer

Attaches OMS PushConsumer to a specified queue and consumes messages by MessageListener

  1. public class OMSPushConsumer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final PushConsumer consumer = messagingAccessPoint.
  7.             createPushConsumer(OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
  9.         messagingAccessPoint.startup();
  10.         System.out.printf("MessagingAccessPoint startup OK%n");
  12.         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 consumer.shutdown();
  16.                 messagingAccessPoint.shutdown();
  17.             }
  18.         }));
  20.         consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() {
  21.             @Override
  22.             public void onMessage(final Message message, final ReceivedMessageContext context) {
  23.                 System.out.printf("Received one message: %s%n", message.headers().getString(MessageHeader.MESSAGE_ID));
  24.                 context.ack();
  25.             }
  26.         });
  28.     }
  29. }

OMSPushConsumer

把OMS PushConsumer和某一个特定的队列绑定在一起，通过MessageListener消费消息。

  1. public class OMSPushConsumer {
  2.     public static void main(String[] args) {
  3.         final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
  4.             .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
  6.         final PushConsumer consumer = messagingAccessPoint.
  7.             createPushConsumer(OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
  9.         messagingAccessPoint.startup();
  10.         System.out.printf("MessagingAccessPoint startup OK%n");
  12.         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 consumer.shutdown();
  16.                 messagingAccessPoint.shutdown();
  17.             }
  18.         }));
  20.         consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() {
  21.             @Override
  22.             public void onMessage(final Message message, final ReceivedMessageContext context) {
  23.                 System.out.printf("Received one message: %s%n", message.headers().getString(MessageHeader.MESSAGE_ID));
  24.                 context.ack();
  25.             }
  26.         });
  28.     }
  29. }