RocketMQ_Example.md 38.5 KB
Newer Older
RocketMQ技术号's avatar
RocketMQ技术号 已提交
1
# 样例
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40
-----
 * [目录](#样例)
      * [1 基本样例](#1-基本样例)
         * [1.1 加入依赖:](#11-加入依赖)
         * [1.2 消息发送](#12-消息发送)
            * [1、Producer端发送同步消息](#1producer端发送同步消息)
            * [2、发送异步消息](#2发送异步消息)
            * [3、单向发送消息](#3单向发送消息)
         * [1.3 消费消息](#13-消费消息)
      * [2 顺序消息样例](#2-顺序消息样例)
         * [2.1 顺序消息生产](#21-顺序消息生产)
         * [2.2 顺序消费消息](#22-顺序消费消息)
      * [3 延时消息样例](#3-延时消息样例)
         * [3.1 启动消费者等待传入订阅消息](#31-启动消费者等待传入订阅消息)
         * [3.2 发送延时消息](#32-发送延时消息)
         * [3.3 验证](#33-验证)
         * [3.4 延时消息的使用场景](#34-延时消息的使用场景)
         * [3.5 延时消息的使用限制](#35-延时消息的使用限制)
      * [4 批量消息样例](#4-批量消息样例)
         * [4.1 发送批量消息](#41-发送批量消息)
         * [4.2 消息列表分割](#42-消息列表分割)
      * [5 过滤消息样例](#5-过滤消息样例)
         * [5.1 基本语法](#51-基本语法)
         * [5.2 使用样例](#52-使用样例)
            * [1、生产者样例](#1生产者样例)
            * [2、消费者样例](#2消费者样例)
      * [6 消息事务样例](#6-消息事务样例)
         * [6.1 发送事务消息样例](#61-发送事务消息样例)
            * [1、创建事务性生产者](#1创建事务性生产者)
            * [2、实现事务的监听接口](#2实现事务的监听接口)
         * [6.2 事务消息使用上的限制](#62-事务消息使用上的限制)
      * [7 Logappender样例](#7-logappender样例)
         * [7.1 log4j样例](#71-log4j样例)
         * [7.2 log4j2样例](#72-log4j2样例)
         * [7.3 logback样例](#73-logback样例)
      * [8 OpenMessaging样例](#8-openmessaging样例)
         * [8.1 OMSProducer样例](#81-omsproducer样例)
         * [8.2 OMSPullConsumer](#82-omspullconsumer)
         * [8.3 OMSPushConsumer](#83-omspushconsumer)
RocketMQ技术号's avatar
RocketMQ技术号 已提交
41 42 43
-----
## 1 基本样例

神奇大叶子's avatar
神奇大叶子 已提交
44 45 46

在基本样例中我们提供如下的功能场景:

47
* 使用RocketMQ发送三种类型的消息:同步消息、异步消息和单向消息。其中前两种消息是可靠的,因为会有发送是否成功的应答。
神奇大叶子's avatar
神奇大叶子 已提交
48 49
* 使用RocketMQ来消费接收到的消息。

神奇大叶子's avatar
神奇大叶子 已提交
50
### 1.1 加入依赖:
神奇大叶子's avatar
神奇大叶子 已提交
51 52 53

`maven:`
```
神奇大叶子's avatar
神奇大叶子 已提交
54
<dependency>
神奇大叶子's avatar
神奇大叶子 已提交
55 56
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
神奇大叶子's avatar
神奇大叶子 已提交
57
    <version>4.3.0</version>
神奇大叶子's avatar
神奇大叶子 已提交
58 59 60 61 62 63
</dependency>
```
`gradle`
```
compile 'org.apache.rocketmq:rocketmq-client:4.3.0'
```
神奇大叶子's avatar
神奇大叶子 已提交
64
### 1.2 消息发送
神奇大叶子's avatar
神奇大叶子 已提交
65

神奇大叶子's avatar
神奇大叶子 已提交
66
#### 1、Producer端发送同步消息
神奇大叶子's avatar
神奇大叶子 已提交
67 68 69 70 71 72 73 74 75 76 77 78

这种可靠性同步地发送方式使用的比较广泛,比如:重要的消息通知,短信通知。
```java
public class SyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
    	producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
神奇大叶子's avatar
神奇大叶子 已提交
79 80
    	    // 创建消息,并指定Topic,Tag和消息体
    	    Message msg = new Message("TopicTest" /* Topic */,
神奇大叶子's avatar
神奇大叶子 已提交
81
        	"TagA" /* Tag */,
神奇大叶子's avatar
神奇大叶子 已提交
82
        	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
神奇大叶子's avatar
神奇大叶子 已提交
83
        	);
神奇大叶子's avatar
神奇大叶子 已提交
84
        	// 发送消息到一个Broker
神奇大叶子's avatar
神奇大叶子 已提交
85 86
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
87
            System.out.printf("%s%n", sendResult);
神奇大叶子's avatar
神奇大叶子 已提交
88
    	}
神奇大叶子's avatar
神奇大叶子 已提交
89
    	// 如果不再发送消息,关闭Producer实例。
90 91
    	producer.shutdown();
    }
神奇大叶子's avatar
神奇大叶子 已提交
92 93
}
```
神奇大叶子's avatar
神奇大叶子 已提交
94
#### 2、发送异步消息
神奇大叶子's avatar
神奇大叶子 已提交
95

神奇大叶子's avatar
神奇大叶子 已提交
96
异步消息通常用在对响应时间敏感的业务场景,即发送端不能容忍长时间地等待Broker的响应。
神奇大叶子's avatar
神奇大叶子 已提交
97 98 99 100 101 102 103 104 105 106 107

```java
public class AsyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
Vhaisong's avatar
Vhaisong 已提交
108 109 110 111 112
	
	int messageCount = 100;
        // 根据消息数量实例化倒计时计算器
	final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
    	for (int i = 0; i < messageCount; i++) {
神奇大叶子's avatar
神奇大叶子 已提交
113
                final int index = i;
神奇大叶子's avatar
神奇大叶子 已提交
114
            	// 创建消息,并指定Topic,Tag和消息体
神奇大叶子's avatar
神奇大叶子 已提交
115 116 117 118 119 120 121 122
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // SendCallback接收异步返回结果的回调
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
123
                        countDownLatch.countDown();
神奇大叶子's avatar
神奇大叶子 已提交
124 125 126 127 128
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
129 130 131
                        countDownLatch.countDown();
      	                System.out.printf("%-10d Exception %s %n", index, e);
      	                e.printStackTrace();
神奇大叶子's avatar
神奇大叶子 已提交
132 133 134
                    }
            	});
    	}
Vhaisong's avatar
Vhaisong 已提交
135 136
	// 等待5s
	countDownLatch.await(5, TimeUnit.SECONDS);
神奇大叶子's avatar
神奇大叶子 已提交
137
    	// 如果不再发送消息,关闭Producer实例。
138 139
    	producer.shutdown();
    }
神奇大叶子's avatar
神奇大叶子 已提交
140 141 142
}
```

神奇大叶子's avatar
神奇大叶子 已提交
143
#### 3、单向发送消息
神奇大叶子's avatar
神奇大叶子 已提交
144

神奇大叶子's avatar
神奇大叶子 已提交
145
这种方式主要用在不特别关心发送结果的场景,例如日志发送。
神奇大叶子's avatar
神奇大叶子 已提交
146 147 148 149 150 151 152 153 154 155 156

```java
public class OnewayProducer {
	public static void main(String[] args) throws Exception{
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
神奇大叶子's avatar
神奇大叶子 已提交
157
        	// 创建消息,并指定Topic,Tag和消息体
神奇大叶子's avatar
神奇大叶子 已提交
158 159 160 161 162
        	Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送单向消息,没有任何返回结果
神奇大叶子's avatar
神奇大叶子 已提交
163
        	producer.sendOneway(msg);
164

神奇大叶子's avatar
神奇大叶子 已提交
165
    	}
神奇大叶子's avatar
神奇大叶子 已提交
166
    	// 如果不再发送消息,关闭Producer实例。
167 168 169
    	producer.shutdown();
    }
}
神奇大叶子's avatar
神奇大叶子 已提交
170 171
```

神奇大叶子's avatar
神奇大叶子 已提交
172
### 1.3 消费消息
神奇大叶子's avatar
神奇大叶子 已提交
173 174 175

```java
public class Consumer {
176

神奇大叶子's avatar
神奇大叶子 已提交
177
	public static void main(String[] args) throws InterruptedException, MQClientException {
178

神奇大叶子's avatar
神奇大叶子 已提交
179 180
    	// 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
181

神奇大叶子's avatar
神奇大叶子 已提交
182 183
    	// 设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
184

神奇大叶子's avatar
神奇大叶子 已提交
185
    	// 订阅一个或者多个Topic,以及Tag来过滤需要消费的消息
神奇大叶子's avatar
神奇大叶子 已提交
186 187 188 189
        consumer.subscribe("TopicTest", "*");
    	// 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
190
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
神奇大叶子's avatar
神奇大叶子 已提交
191
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
192
                // 标记该消息已经被成功消费
神奇大叶子's avatar
神奇大叶子 已提交
193
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
194 195 196
            }
        });
        // 启动消费者实例
神奇大叶子's avatar
神奇大叶子 已提交
197 198 199 200 201 202
        consumer.start();
        System.out.printf("Consumer Started.%n");
	}
}
```

神奇大叶子's avatar
神奇大叶子 已提交
203
2 顺序消息样例
神奇大叶子's avatar
神奇大叶子 已提交
204 205
----------

神奇大叶子's avatar
神奇大叶子 已提交
206
消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序,可以分为分区有序或者全局有序。
神奇大叶子's avatar
神奇大叶子 已提交
207

神奇大叶子's avatar
神奇大叶子 已提交
208
顺序消费的原理解析,在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列);而消费消息的时候从多个queue上拉取消息,这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中,消费的时候只从这个queue上依次拉取,则就保证了顺序。当发送和消费参与的queue只有一个,则是全局有序;如果多个queue参与,则为分区有序,即相对每个queue,消息都是有序的。
神奇大叶子's avatar
神奇大叶子 已提交
209

210
下面用订单进行分区有序的示例。一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中,消费时,同一个OrderId获取到的肯定是同一个队列。
神奇大叶子's avatar
神奇大叶子 已提交
211

神奇大叶子's avatar
神奇大叶子 已提交
212
### 2.1 顺序消息生产
神奇大叶子's avatar
神奇大叶子 已提交
213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250

```java
package org.apache.rocketmq.example.order2;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
* Producer,发送顺序消息
*/
public class Producer {

   public static void main(String[] args) throws Exception {
       DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");

       producer.setNamesrvAddr("127.0.0.1:9876");

       producer.start();

       String[] tags = new String[]{"TagA", "TagC", "TagD"};

       // 订单列表
       List<OrderStep> orderList = new Producer().buildOrders();

       Date date = new Date();
       SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
       String dateStr = sdf.format(date);
       for (int i = 0; i < 10; i++) {
           // 加个时间前缀
           String body = dateStr + " Hello RocketMQ " + orderList.get(i);
神奇大叶子's avatar
神奇大叶子 已提交
251
           Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
神奇大叶子's avatar
神奇大叶子 已提交
252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363

           SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
               @Override
               public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                   Long id = (Long) arg;  //根据订单id选择发送queue
                   long index = id % mqs.size();
                   return mqs.get((int) index);
               }
           }, orderList.get(i).getOrderId());//订单id

           System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
               sendResult.getSendStatus(),
               sendResult.getMessageQueue().getQueueId(),
               body));
       }

       producer.shutdown();
   }

   /**
    * 订单的步骤
    */
   private static class OrderStep {
       private long orderId;
       private String desc;

       public long getOrderId() {
           return orderId;
       }

       public void setOrderId(long orderId) {
           this.orderId = orderId;
       }

       public String getDesc() {
           return desc;
       }

       public void setDesc(String desc) {
           this.desc = desc;
       }

       @Override
       public String toString() {
           return "OrderStep{" +
               "orderId=" + orderId +
               ", desc='" + desc + '\'' +
               '}';
       }
   }

   /**
    * 生成模拟订单数据
    */
   private List<OrderStep> buildOrders() {
       List<OrderStep> orderList = new ArrayList<OrderStep>();

       OrderStep orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("推送");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       return orderList;
   }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
364
### 2.2 顺序消费消息
神奇大叶子's avatar
神奇大叶子 已提交
365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393

```java
package org.apache.rocketmq.example.order2;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
* 顺序消息消费,带事务方式(应用可控制Offset什么时候提交)
*/
public class ConsumerInOrder {

   public static void main(String[] args) throws Exception {
       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
       consumer.setNamesrvAddr("127.0.0.1:9876");
       /**
        * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
        * 如果非第一次启动,那么按照上次消费的位置继续消费
        */
       consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

神奇大叶子's avatar
神奇大叶子 已提交
394
       consumer.subscribe("TopicTest", "TagA || TagC || TagD");
神奇大叶子's avatar
神奇大叶子 已提交
395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424

       consumer.registerMessageListener(new MessageListenerOrderly() {

           Random random = new Random();

           @Override
           public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
               context.setAutoCommit(true);
               for (MessageExt msg : msgs) {
                   // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                   System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
               }

               try {
                   //模拟业务逻辑处理中...
                   TimeUnit.SECONDS.sleep(random.nextInt(10));
               } catch (Exception e) {
                   e.printStackTrace();
               }
               return ConsumeOrderlyStatus.SUCCESS;
           }
       });

       consumer.start();

       System.out.println("Consumer Started.");
   }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
425
3 延时消息样例
神奇大叶子's avatar
神奇大叶子 已提交
426 427
----------

神奇大叶子's avatar
神奇大叶子 已提交
428
### 3.1 启动消费者等待传入订阅消息
神奇大叶子's avatar
神奇大叶子 已提交
429 430 431 432 433 434 435 436 437 438 439 440 441 442

```java

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class ScheduledMessageConsumer {
   public static void main(String[] args) throws Exception {
      // 实例化消费者
      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
神奇大叶子's avatar
神奇大叶子 已提交
443
      // 订阅Topics
神奇大叶子's avatar
神奇大叶子 已提交
444 445 446 447 448 449 450
      consumer.subscribe("TestTopic", "*");
      // 注册消息监听者
      consumer.registerMessageListener(new MessageListenerConcurrently() {
          @Override
          public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
              for (MessageExt message : messages) {
                  // Print approximate delay time period
451
                  System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getBornTimestamp()) + "ms later");
神奇大叶子's avatar
神奇大叶子 已提交
452 453 454 455 456 457 458 459 460 461 462
              }
              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
      });
      // 启动消费者
      consumer.start();
  }
}

```

神奇大叶子's avatar
神奇大叶子 已提交
463
### 3.2 发送延时消息
神奇大叶子's avatar
神奇大叶子 已提交
464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489

```java

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class ScheduledMessageProducer {
   public static void main(String[] args) throws Exception {
      // 实例化一个生产者来产生延时消息
      DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
      // 启动生产者
      producer.start();
      int totalMessagesToSend = 100;
      for (int i = 0; i < totalMessagesToSend; i++) {
          Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
          // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
          message.setDelayTimeLevel(3);
          // 发送消息
          producer.send(message);
      }
       // 关闭生产者
      producer.shutdown();
  }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
490
### 3.3 验证
神奇大叶子's avatar
神奇大叶子 已提交
491 492 493

您将会看到消息的消费比存储时间晚10秒。

神奇大叶子's avatar
神奇大叶子 已提交
494
### 3.4 延时消息的使用场景
495
比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单的状态,如果还是未付款就取消订单释放库存。
496

神奇大叶子's avatar
神奇大叶子 已提交
497
### 3.5 延时消息的使用限制
498 499 500 501 502 503 504 505 506 507

```java
// org/apache/rocketmq/store/config/MessageStoreConfig.java

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
```
现在RocketMq并不支持任意时间的延时,需要设置几个固定的延时等级,从1s到2h分别对应着等级1到18
消息消费失败会进入延时消息队列,消息发送时间与设置的延时等级和重试次数有关,详见代码`SendMessageProcessor.java`


神奇大叶子's avatar
神奇大叶子 已提交
508
4 批量消息样例
神奇大叶子's avatar
神奇大叶子 已提交
509 510
----------

神奇大叶子's avatar
神奇大叶子 已提交
511
批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic,相同的waitStoreMsgOK,而且不能是延时消息。此外,这一批消息的总大小不应超过4MB。
神奇大叶子's avatar
神奇大叶子 已提交
512

神奇大叶子's avatar
神奇大叶子 已提交
513
### 4.1 发送批量消息
神奇大叶子's avatar
神奇大叶子 已提交
514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531

如果您每次只发送不超过4MB的消息,则很容易使用批处理,样例如下:

```java
String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {
   producer.send(messages);
} catch (Exception e) {
   e.printStackTrace();
   //处理error
}

```

神奇大叶子's avatar
神奇大叶子 已提交
532
### 4.2 消息列表分割
神奇大叶子's avatar
神奇大叶子 已提交
533 534 535 536

复杂度只有当你发送大批量时才会增长,你可能不确定它是否超过了大小限制(4MB)。这时候你最好把你的消息列表分割一下:

```java
537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582
public class ListSplitter implements Iterator<List<Message>> { 
    private final int SIZE_LIMIT = 1024 * 1024 * 4;
    private final List<Message> messages;
    private int currIndex;
    public ListSplitter(List<Message> messages) { 
        this.messages = messages;
    }
    @Override public boolean hasNext() {
        return currIndex < messages.size(); 
    }
    @Override public List<Message> next() { 
        int startIndex = getStartIndex();
        int nextIndex = startIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex); 
            int tmpSize = calcMessageSize(message);
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break; 
            } else {
                totalSize += tmpSize; 
            }
        }
        List<Message> subList = messages.subList(startIndex, nextIndex); 
        currIndex = nextIndex;
        return subList;
    }
    private int getStartIndex() {
        Message currMessage = messages.get(currIndex); 
        int tmpSize = calcMessageSize(currMessage); 
        while(tmpSize > SIZE_LIMIT) {
            currIndex += 1;
            Message message = messages.get(curIndex); 
            tmpSize = calcMessageSize(message);
        }
        return currIndex; 
    }
    private int calcMessageSize(Message message) {
        int tmpSize = message.getTopic().length() + message.getBody().length(); 
        Map<String, String> properties = message.getProperties();
        for (Map.Entry<String, String> entry : properties.entrySet()) {
            tmpSize += entry.getKey().length() + entry.getValue().length(); 
        }
        tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
        return tmpSize; 
    }
神奇大叶子's avatar
神奇大叶子 已提交
583 584 585 586 587 588 589 590 591 592 593 594 595 596
}
//把大的消息分裂成若干个小的消息
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
  try {
      List<Message>  listItem = splitter.next();
      producer.send(listItem);
  } catch (Exception e) {
      e.printStackTrace();
      //处理error
  }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
597
5 过滤消息样例
神奇大叶子's avatar
神奇大叶子 已提交
598 599 600 601 602 603 604 605 606 607
----------

在大多数情况下,TAG是一个简单而有用的设计,其可以来选择您想要的消息。例如:

```java
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");
```

消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签,这对于复杂的场景可能不起作用。在这种情况下,可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下,可以实现一些简单的逻辑。下面是一个例子:
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623
```
------------
| message  |
|----------|  a > 5 AND b = 'abc'
| a = 10   |  --------------------> Gotten
| b = 'abc'|
| c = true |
------------
------------
| message  |
|----------|   a > 5 AND b = 'abc'
| a = 1    |  --------------------> Missed
| b = 'abc'|
| c = true |
------------
```
神奇大叶子's avatar
神奇大叶子 已提交
624
### 5.1 基本语法
神奇大叶子's avatar
神奇大叶子 已提交
625 626 627

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

628 629 630 631
- 数值比较,比如:**>,>=,<,<=,BETWEEN,=;**
- 字符比较,比如:**=,<>,IN;**
- **IS NULL** 或者 **IS NOT NULL;**
- 逻辑符号 **AND,OR,NOT;**
神奇大叶子's avatar
神奇大叶子 已提交
632 633 634

常量支持类型为:

635 636 637 638
- 数值,比如:**123,3.1415;**
- 字符,比如:**'abc',必须用单引号包裹起来;**
- **NULL**,特殊的常量
- 布尔值,**TRUE** 或 **FALSE**
神奇大叶子's avatar
神奇大叶子 已提交
639 640 641

只有使用push模式的消费者才能用使用SQL92标准的sql语句,接口如下:
```
642
public void subscribe(finalString topic, final MessageSelector messageSelector)
神奇大叶子's avatar
神奇大叶子 已提交
643 644
```

神奇大叶子's avatar
神奇大叶子 已提交
645
### 5.2 使用样例
神奇大叶子's avatar
神奇大叶子 已提交
646

神奇大叶子's avatar
神奇大叶子 已提交
647
#### 1、生产者样例
神奇大叶子's avatar
神奇大叶子 已提交
648

神奇大叶子's avatar
神奇大叶子 已提交
649
发送消息时,你能通过`putUserProperty`来设置消息的属性
神奇大叶子's avatar
神奇大叶子 已提交
650 651 652 653 654 655 656 657 658 659 660

```java
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
Message msg = new Message("TopicTest",
   tag,
   ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// 设置一些属性
msg.putUserProperty("a", String.valueOf(i));
SendResult sendResult = producer.send(msg);
661

神奇大叶子's avatar
神奇大叶子 已提交
662 663 664
producer.shutdown();
```

神奇大叶子's avatar
神奇大叶子 已提交
665
#### 2、消费者样例
神奇大叶子's avatar
神奇大叶子 已提交
666

神奇大叶子's avatar
神奇大叶子 已提交
667
用MessageSelector.bySql来使用sql筛选消息
神奇大叶子's avatar
神奇大叶子 已提交
668 669 670 671 672 673 674 675 676 677 678 679 680 681 682

```java
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
// 只有订阅的消息有这个属性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {
   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }
});
consumer.start();

```

神奇大叶子's avatar
神奇大叶子 已提交
683
6 消息事务样例
神奇大叶子's avatar
神奇大叶子 已提交
684 685 686 687
----------

事务消息共有三种状态,提交状态、回滚状态、中间状态:

688 689 690
- TransactionStatus.CommitTransaction: 提交事务,它允许消费者消费此消息。
- TransactionStatus.RollbackTransaction: 回滚事务,它代表该消息将被删除,不允许被消费。
- TransactionStatus.Unknown: 中间状态,它代表需要检查消息队列来确定状态。
神奇大叶子's avatar
神奇大叶子 已提交
691

神奇大叶子's avatar
神奇大叶子 已提交
692
### 6.1 发送事务消息样例
神奇大叶子's avatar
神奇大叶子 已提交
693

神奇大叶子's avatar
神奇大叶子 已提交
694
#### 1、创建事务性生产者
神奇大叶子's avatar
神奇大叶子 已提交
695

696
使用 `TransactionMQProducer`类创建生产者,并指定唯一的 `ProducerGroup`,就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。
神奇大叶子's avatar
神奇大叶子 已提交
697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740

```java
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class TransactionProducer {
   public static void main(String[] args) throws MQClientException, InterruptedException {
       TransactionListener transactionListener = new TransactionListenerImpl();
       TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
       ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread thread = new Thread(r);
               thread.setName("client-transaction-msg-check-thread");
               return thread;
           }
       });
       producer.setExecutorService(executorService);
       producer.setTransactionListener(transactionListener);
       producer.start();
       String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
       for (int i = 0; i < 10; i++) {
           try {
               Message msg =
                   new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                       ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
               SendResult sendResult = producer.sendMessageInTransaction(msg, null);
               System.out.printf("%s%n", sendResult);
               Thread.sleep(10);
           } catch (MQClientException | UnsupportedEncodingException e) {
               e.printStackTrace();
           }
       }
       for (int i = 0; i < 100000; i++) {
           Thread.sleep(1000);
       }
       producer.shutdown();
   }
}

```
神奇大叶子's avatar
神奇大叶子 已提交
741
#### 2、实现事务的监听接口
神奇大叶子's avatar
神奇大叶子 已提交
742

H
H 已提交
743
当发送半消息成功时,我们使用 `executeLocalTransaction` 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。`checkLocalTransaction` 方法用于检查本地事务状态,并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。
神奇大叶子's avatar
神奇大叶子 已提交
744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774

```java
public class TransactionListenerImpl implements TransactionListener {
  private AtomicInteger transactionIndex = new AtomicInteger(0);
  private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
  @Override
  public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
      int value = transactionIndex.getAndIncrement();
      int status = value % 3;
      localTrans.put(msg.getTransactionId(), status);
      return LocalTransactionState.UNKNOW;
  }
  @Override
  public LocalTransactionState checkLocalTransaction(MessageExt msg) {
      Integer status = localTrans.get(msg.getTransactionId());
      if (null != status) {
          switch (status) {
              case 0:
                  return LocalTransactionState.UNKNOW;
              case 1:
                  return LocalTransactionState.COMMIT_MESSAGE;
              case 2:
                  return LocalTransactionState.ROLLBACK_MESSAGE;
          }
      }
      return LocalTransactionState.COMMIT_MESSAGE;
  }
}

```

神奇大叶子's avatar
神奇大叶子 已提交
775
### 6.2 事务消息使用上的限制
神奇大叶子's avatar
神奇大叶子 已提交
776 777

1. 事务消息不支持延时消息和批量消息。
A
andy.xu 已提交
778 779
2. 为了避免单个消息被检查太多次而导致半队列消息累积,我们默认将单个消息的检查次数限制为 15 次,但是用户可以通过 Broker 配置文件的 `transactionCheckMax`参数来修改此限制。如果已经检查某条消息超过 N 次的话( N = `transactionCheckMax` ) 则 Broker 将丢弃此消息,并在默认情况下同时打印错误日志。用户可以通过重写 `AbstractTransactionalMessageCheckListener` 类来修改这个行为。
3. 事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时,用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制,该参数优先于 `transactionTimeout` 参数。
神奇大叶子's avatar
神奇大叶子 已提交
780 781
4. 事务性消息可能不止一次被检查或消费。
5. 提交给用户的目标主题消息可能会失败,目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证,如果希望确保事务消息不丢失、并且事务完整性得到保证,建议使用同步的双重写入机制。
782
6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同,事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。
神奇大叶子's avatar
神奇大叶子 已提交
783

神奇大叶子's avatar
神奇大叶子 已提交
784
7 Logappender样例
神奇大叶子's avatar
神奇大叶子 已提交
785 786 787 788
-----------------

RocketMQ日志提供log4j、log4j2和logback日志框架作为业务应用,下面是配置样例

神奇大叶子's avatar
神奇大叶子 已提交
789
### 7.1 log4j样例
神奇大叶子's avatar
神奇大叶子 已提交
790 791 792

按下面样例使用log4j属性配置
```
神奇大叶子's avatar
神奇大叶子 已提交
793 794 795 796 797 798
log4j.appender.mq=org.apache.rocketmq.logappender.log4j.RocketmqLog4jAppender
log4j.appender.mq.Tag=yourTag
log4j.appender.mq.Topic=yourLogTopic
log4j.appender.mq.ProducerGroup=yourLogGroup
log4j.appender.mq.NameServerAddress=yourRocketmqNameserverAddress
log4j.appender.mq.layout=org.apache.log4j.PatternLayout
神奇大叶子's avatar
神奇大叶子 已提交
799 800 801 802
log4j.appender.mq.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-4r [%t] (%F:%L) %-5p - %m%n
```
按下面样例使用log4j xml配置来使用异步添加日志
```
神奇大叶子's avatar
神奇大叶子 已提交
803 804 805 806 807 808 809 810 811 812 813 814 815
<appender name="mqAppender1"class="org.apache.rocketmq.logappender.log4j.RocketmqLog4jAppender">
  <param name="Tag" value="yourTag" />
  <param name="Topic" value="yourLogTopic" />
  <param name="ProducerGroup" value="yourLogGroup" />
  <param name="NameServerAddress" value="yourRocketmqNameserverAddress"/>
  <layout class="org.apache.log4j.PatternLayout">
      <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss}-%p %t %c - %m%n" />
  </layout>
</appender>
<appender name="mqAsyncAppender1"class="org.apache.log4j.AsyncAppender">
  <param name="BufferSize" value="1024" />
  <param name="Blocking" value="false" />
  <appender-ref ref="mqAppender1"/>
神奇大叶子's avatar
神奇大叶子 已提交
816 817
</appender>
```
神奇大叶子's avatar
神奇大叶子 已提交
818
### 7.2 log4j2样例
神奇大叶子's avatar
神奇大叶子 已提交
819 820 821

用log4j2时,配置如下,如果想要非阻塞,只需要使用异步添加引用即可
```
神奇大叶子's avatar
神奇大叶子 已提交
822 823 824
<RocketMQ name="rocketmqAppender" producerGroup="yourLogGroup" nameServerAddress="yourRocketmqNameserverAddress"
   topic="yourLogTopic" tag="yourTag">
  <PatternLayout pattern="%d [%p] hahahah %c %m%n"/>
神奇大叶子's avatar
神奇大叶子 已提交
825 826
</RocketMQ>
```
神奇大叶子's avatar
神奇大叶子 已提交
827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847
### 7.3 logback样例
```
<appender name="mqAppender1"class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">
  <tag>yourTag</tag>
  <topic>yourLogTopic</topic>
  <producerGroup>yourLogGroup</producerGroup>
  <nameServerAddress>yourRocketmqNameserverAddress</nameServerAddress>
  <layout>
      <pattern>%date %p %t - %m%n</pattern>
  </layout>
</appender>
<appender name="mqAsyncAppender1"class="ch.qos.logback.classic.AsyncAppender">
  <queueSize>1024</queueSize>
  <discardingThreshold>80</discardingThreshold>
  <maxFlushTime>2000</maxFlushTime>
  <neverBlock>true</neverBlock>
  <appender-ref ref="mqAppender1"/>
</appender>
```

8 OpenMessaging样例
神奇大叶子's avatar
神奇大叶子 已提交
848
---------------
神奇大叶子's avatar
神奇大叶子 已提交
849 850 851

 [OpenMessaging](https://www.google.com/url?q=http://openmessaging.cloud/&sa=D&ust=1546524111089000)旨在建立消息和流处理规范,以为金融、电子商务、物联网和大数据领域提供通用框架及工业级指导方案。在分布式异构环境中,设计原则是面向云、简单、灵活和独立于语言。符合这些规范将帮助企业方便的开发跨平台和操作系统的异构消息传递应用程序。提供了openmessaging-api 0.3.0-alpha的部分实现,下面的示例演示如何基于OpenMessaging访问RocketMQ。

神奇大叶子's avatar
神奇大叶子 已提交
852
### 8.1 OMSProducer样例
神奇大叶子's avatar
神奇大叶子 已提交
853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910

下面的示例演示如何在同步、异步或单向传输中向RocketMQ代理发送消息。

```java
import io.openmessaging.Future;
import io.openmessaging.FutureListener;
import io.openmessaging.Message;
import io.openmessaging.MessagingAccessPoint;
import io.openmessaging.OMS;
import io.openmessaging.producer.Producer;
import io.openmessaging.producer.SendResult;
import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;

public class SimpleProducer {
    public static void main(String[] args) {
       final MessagingAccessPoint messagingAccessPoint =
           OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
       final Producer producer = messagingAccessPoint.createProducer();
       messagingAccessPoint.startup();
       System.out.printf("MessagingAccessPoint startup OK%n");
       producer.startup();
       System.out.printf("Producer startup OK%n");
       {
           Message message = producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
           SendResult sendResult = producer.send(message);
           //final Void aVoid = result.get(3000L);
           System.out.printf("Send async message OK, msgId: %s%n", sendResult.messageId());
       }
       final CountDownLatch countDownLatch = new CountDownLatch(1);
       {
           final Future<SendResult> result = producer.sendAsync(producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
           result.addListener(new FutureListener<SendResult>() {
               @Override
               public void operationComplete(Future<SendResult> future) {
                   if (future.getThrowable() != null) {
                       System.out.printf("Send async message Failed, error: %s%n", future.getThrowable().getMessage());
                   } else {
                       System.out.printf("Send async message OK, msgId: %s%n", future.get().messageId());
                   }
                   countDownLatch.countDown();
               }
           });
       }
       {
           producer.sendOneway(producer.createBytesMessage("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
           System.out.printf("Send oneway message OK%n");
       }
       try {
           countDownLatch.await();
           Thread.sleep(500); // 等一些时间来发送消息
       } catch (InterruptedException ignore) {
       }
       producer.shutdown();
   }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
911
### 8.2 OMSPullConsumer
神奇大叶子's avatar
神奇大叶子 已提交
912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963

用OMS PullConsumer 来从指定的队列中拉取消息

```java
import io.openmessaging.Message;
import io.openmessaging.MessagingAccessPoint;
import io.openmessaging.OMS;
import io.openmessaging.OMSBuiltinKeys;
import io.openmessaging.consumer.PullConsumer;
import io.openmessaging.producer.Producer;
import io.openmessaging.producer.SendResult;

public class SimplePullConsumer {
    public static void main(String[] args) {
       final MessagingAccessPoint messagingAccessPoint =
           OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
       messagingAccessPoint.startup();
       final Producer producer = messagingAccessPoint.createProducer();
       final PullConsumer consumer = messagingAccessPoint.createPullConsumer(
           OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
       messagingAccessPoint.startup();
       System.out.printf("MessagingAccessPoint startup OK%n");
       final String queueName = "TopicTest";
       producer.startup();
       Message msg = producer.createBytesMessage(queueName, "Hello Open Messaging".getBytes());
       SendResult sendResult = producer.send(msg);
       System.out.printf("Send Message OK. MsgId: %s%n", sendResult.messageId());
       producer.shutdown();
       consumer.attachQueue(queueName);
       consumer.startup();
       System.out.printf("Consumer startup OK%n");
       // 运行直到发现一个消息被发送了
       boolean stop = false;
       while (!stop) {
           Message message = consumer.receive();
           if (message != null) {
               String msgId = message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID);
               System.out.printf("Received one message: %s%n", msgId);
               consumer.ack(msgId);
               if (!stop) {
                   stop = msgId.equalsIgnoreCase(sendResult.messageId());
               }
           } else {
               System.out.printf("Return without any message%n");
           }
       }
       consumer.shutdown();
       messagingAccessPoint.shutdown();
   }
}
```

神奇大叶子's avatar
神奇大叶子 已提交
964
### 8.3 OMSPushConsumer
神奇大叶子's avatar
神奇大叶子 已提交
965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001

以下示范如何将 OMS PushConsumer 添加到指定的队列,并通过 MessageListener 消费这些消息。

```java
import io.openmessaging.Message;
import io.openmessaging.MessagingAccessPoint;
import io.openmessaging.OMS;
import io.openmessaging.OMSBuiltinKeys;
import io.openmessaging.consumer.MessageListener;
import io.openmessaging.consumer.PushConsumer;

public class SimplePushConsumer {
    public static void main(String[] args) {
       final MessagingAccessPoint messagingAccessPoint = OMS
           .getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
       final PushConsumer consumer = messagingAccessPoint.
           createPushConsumer(OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
       messagingAccessPoint.startup();
       System.out.printf("MessagingAccessPoint startup OK%n");
       Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
           @Override
           public void run() {
               consumer.shutdown();
               messagingAccessPoint.shutdown();
           }
       }));
       consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() {
           @Override
           public void onReceived(Message message, Context context) {
               System.out.printf("Received one message: %s%n", message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID));
               context.ack();
           }
       });
       consumer.startup();
       System.out.printf("Consumer startup OK%n");
   }
}
RocketMQ技术号's avatar
RocketMQ技术号 已提交
1002
```