安科网

RocketMQ生产者消息篇

Qemo

Qemo

2019-07-01

关注 关注

系列文章

RocketMQ入门篇
RocketMQ生产者流程篇
RocketMQ生产者消息篇

前言

上文RocketMQ生产者流程篇中详细介绍了生产者发送消息的流程，本文将重点介绍发送消息的通信模式以及各种不同的消息类型。

通信模式

RocketMQ提供了三种通讯模式，分别是：同步，异步和单向；可以查看内部类CommunicationMode：

public enum CommunicationMode {
    SYNC,
    ASYNC,
    ONEWAY,
}

下面分别看一下三种通讯模式如何使用

1.同步方式

看一个简单的发送同步消息的实例：

public class SyncProducer {

    public static void main(String[] args) throws Exception {

        System.setProperty("rocketmq.namesrv.domain", "localhost");
        // 构造Producer
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName2");
        // producer.setNamesrvAddr("192.168.237.128:9876");
        // 初始化Producer，整个应用生命周期内，只需要初始化1次
        producer.start();

        for (int i = 0; i < 1; i++) {
            Message msg = new Message("TopicTest6", "TagA",
                    ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.println(sendResult);
        }
        producer.shutdown();
    }
}

最简单的直接指定一个message参数默认使用的就是同步方式发送消息，可以看到在发送完消息之后，会立马返回了发送结果SendResult:

SendResult [sendStatus=SEND_OK, msgId=0A0D5307324873D16E9365360AC60000, offsetMsgId=0A0D530700002A9F0000000000001200, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=0]

2.异步方式

看一个简单的发送异步消息的实例：

public class AsyncProducer {

    public static void main(String[] args)
            throws MQClientException, RemotingException, InterruptedException, UnsupportedEncodingException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName2");
        producer.setRetryTimesWhenSendAsyncFailed(3);
        producer.setNamesrvAddr("192.168.237.128:9876");
        producer.start();
        for (int i = 0; i < 1; i++) {
            Message msg = new Message("TopicTest6", "TagA",
                    ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

可以看到在发送消息时指定了SendCallback回调类，send发送方法返回值为void，发送成功之后会回调SendCallback的onSuccess方法，异常调用onException方法；发送成功日志如下：

SendResult [sendStatus=SEND_OK, msgId=0A0D5307261473D16E936536591E0000, offsetMsgId=0A0D530700002A9F00000000000012B2, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=1]

3.单向方式

看一个简单的发送单向消息的实例：

public class OneWayProducer {

    public static void main(String[] args) throws Exception {

        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName2");
        producer.setNamesrvAddr("192.168.237.128:9876");
        producer.start();
        for (int i = 0; i < 1; i++) {
            Message msg = new Message("TopicTest6", "TagA",
                    ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            producer.sendOneway(msg);
        }
        producer.shutdown();
    }
}

单向发送消息发送之后没有响应，但是在消费端可以收到消息，如下所示：

ConsumeMessageThread_6Receive New Messages :[MessageExt [queueId=2, storeSize=178, queueOffset=0, sysFlag=0, bornTimestamp=1550648529526, bornHost=/10.13.83.7:54213, storeTimestamp=1550648529530, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000001416, commitLogOffset=5142, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, CONSUME_START_TIME=1550648529533, UNIQ_KEY=0A0D530747DC73D16E93653766750000, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]

4.发送状态

以上同步和异步实例显示的发送状态都是SEND_OK，除了此状态还有其他三个状态：FLUSH_DISK_TIMEOUT，FLUSH_SLAVE_TIMEOUT和SLAVE_NOT_AVAILABLE；具体可以查看内部类SendStatus：

public enum SendStatus {
    SEND_OK,
    FLUSH_DISK_TIMEOUT,
    FLUSH_SLAVE_TIMEOUT,
    SLAVE_NOT_AVAILABLE,
}

FLUSH_DISK_TIMEOUT：刷盘超时，Broker设置的刷盘策略为SYNC_FLUSH才可能出现此错误；
FLUSH_SLAVE_TIMEOUT：主从同步超时，Broker设置了slave，并且指定同步策略为SYNC_Master；
SLAVE_NOT_AVAILABLE：找不到salve，同样是Broke指定同步策略为SYNC_Master；
SEND_OK：表示发送成功，以上情况都没有出现。
注：必要时需要对各种异常场景进行处理，保证高质量的生产者。

消息类型

在RocketMQ的生产者端可以发送多种类型的消息包括：延迟消息，顺序消息以及事务消息，下面分别进行实例分析；

1.延迟消息

RocketMQ支持发送延迟消息，Broker收到消息后会延迟一段时间在处理，具体使用看如下代码：

Message msg = new Message("TopicTest6", "TagA",
                    ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.setDelayTimeLevel(3);

可以直接设置延迟时间等级，具体有哪些等级，以及每个等级对应的时间可以查看类MessageStoreConfig

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

类中的messageDelayLevel 变量包含了所有可以延迟的时间，使用空格分离，所以这里等级为3，其实对应的延迟时间就是10秒；分别观察生产者和消费者的日志如下：
生产者发送消息日志如下：

Time [Wed Feb 20 18:25:32 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307268073D16E9365CCFA9E0000, offsetMsgId=0A0D530700002A9F0000000000001F42, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=1], queueOffset=4]

消费者接收日志如下：

Time [Wed Feb 20 18:25:42 CST 2019],ConsumeMessageThread_3Receive New Messages :[MessageExt [queueId=1, storeSize=219, queueOffset=2, sysFlag=0, bornTimestamp=1550658332318, bornHost=/10.13.83.7:55682, storeTimestamp=1550658342322, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002026, commitLogOffset=8230, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=3, CONSUME_START_TIME=1550658342325, UNIQ_KEY=0A0D5307268073D16E9365CCFA9E0000, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=1}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]

可以发现发送消息的时间和接收消息的时间相差10秒；

2.顺序消息

顺序消息指生产者生产数据的顺序和消费者消费数据的顺序是一致的；顺序消息包括全局顺序消息和局部顺序消息，全局顺序消息指在某个Topic下所有消息都是顺序的，局部顺序消息指在Topic下的Message Queue中是顺序的；

2.1全局顺序消息

RocketMQ在默认情况下并不能保证有序，一个Topic下会指定多个读写队列，生产者会将消息写入任意的Message Queue中，同样消费者可能会启动多个线程同时处理数据，所以并不能保证顺序；如何保证全局顺序需要只有一个读队列一个写队列，然后需要保证生产者和消费者不能并发处理，以下做一个简单的实例验证；
生产者顺序的发送5条数据：

Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_4Receive New Messages :[MessageExt [queueId=1, storeSize=219, queueOffset=3, sysFlag=0, bornTimestamp=1550713381399, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391403, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002575, commitLogOffset=9589, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=4, CONSUME_START_TIME=1550713391405, UNIQ_KEY=0A0D5307418073D16E936914F6160000, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=1}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_5Receive New Messages :[MessageExt [queueId=2, storeSize=219, queueOffset=2, sysFlag=0, bornTimestamp=1550713381450, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391451, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002650, commitLogOffset=9808, bodyCRC=1561245975, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=3, CONSUME_START_TIME=1550713391453, UNIQ_KEY=0A0D5307418073D16E936914F64A0001, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=2}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 49], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_6Receive New Messages :[MessageExt [queueId=3, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381455, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391460, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F000000000000272B, commitLogOffset=10027, bodyCRC=1141369005, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391463, UNIQ_KEY=0A0D5307418073D16E936914F64F0002, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=3}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 50], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_7Receive New Messages :[MessageExt [queueId=0, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381462, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391464, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002806, commitLogOffset=10246, bodyCRC=855693371, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391490, UNIQ_KEY=0A0D5307418073D16E936914F6560003, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=0}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 51], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_8Receive New Messages :[MessageExt [queueId=1, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381466, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391469, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000028E1, commitLogOffset=10465, bodyCRC=761548184, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391501, UNIQ_KEY=0A0D5307418073D16E936914F65A0004, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=1}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 52], transactionId='null'}]]

可以发现这5条数据分别写入了4个Message Queue中；再看一下消费者日志：

Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_4Receive New Messages :[MessageExt [queueId=1, storeSize=219, queueOffset=3, sysFlag=0, bornTimestamp=1550713381399, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391403, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002575, commitLogOffset=9589, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=4, CONSUME_START_TIME=1550713391405, UNIQ_KEY=0A0D5307418073D16E936914F6160000, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=1}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_5Receive New Messages :[MessageExt [queueId=2, storeSize=219, queueOffset=2, sysFlag=0, bornTimestamp=1550713381450, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391451, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002650, commitLogOffset=9808, bodyCRC=1561245975, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=3, CONSUME_START_TIME=1550713391453, UNIQ_KEY=0A0D5307418073D16E936914F64A0001, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=2}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 49], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_6Receive New Messages :[MessageExt [queueId=3, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381455, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391460, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F000000000000272B, commitLogOffset=10027, bodyCRC=1141369005, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391463, UNIQ_KEY=0A0D5307418073D16E936914F64F0002, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=3}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 50], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_7Receive New Messages :[MessageExt [queueId=0, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381462, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391464, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002806, commitLogOffset=10246, bodyCRC=855693371, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391490, UNIQ_KEY=0A0D5307418073D16E936914F6560003, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=0}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 51], transactionId='null'}]]
Time [Thu Feb 21 09:43:11 CST 2019],ConsumeMessageThread_8Receive New Messages :[MessageExt [queueId=1, storeSize=219, queueOffset=4, sysFlag=0, bornTimestamp=1550713381466, bornHost=/10.13.83.7:55720, storeTimestamp=1550713391469, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000028E1, commitLogOffset=10465, bodyCRC=761548184, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest6, MAX_OFFSET=5, CONSUME_START_TIME=1550713391501, UNIQ_KEY=0A0D5307418073D16E936914F65A0004, WAIT=true, DELAY=3, TAGS=TagA, REAL_QID=1}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 52], transactionId='null'}]]

消费者启动了5个线程同时从4个Message Queue中读取数据，所有不能保证数据的顺序性；
分别做如下改造：设置Topic的读写队列分别为1，可以直接去RocketMQ-console去修改配置；然后设置消费者的处理线程数为1：

consumer.setConsumeThreadMin(1);
consumer.setConsumeThreadMax(1);

再次测试，生产者同样发送5条数据：

Time [Thu Feb 21 09:59:49 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D53071A0873D16E93692457700000, offsetMsgId=0A0D530700002A9F0000000000002EF5, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=11]
Time [Thu Feb 21 09:59:49 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D53071A0873D16E93692457790001, offsetMsgId=0A0D530700002A9F0000000000002FA7, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=12]
Time [Thu Feb 21 09:59:49 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D53071A0873D16E936924577F0002, offsetMsgId=0A0D530700002A9F0000000000003059, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=13]
Time [Thu Feb 21 09:59:49 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D53071A0873D16E93692457850003, offsetMsgId=0A0D530700002A9F000000000000310B, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=14]
Time [Thu Feb 21 09:59:49 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D53071A0873D16E93692457880004, offsetMsgId=0A0D530700002A9F00000000000031BD, messageQueue=MessageQueue [topic=TopicTest6, brokerName=broker-a, queueId=0], queueOffset=15]

可以发送所有的消息都写入了相同的队列，然后看消费者日志：

Time [Thu Feb 21 09:59:49 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=11, sysFlag=0, bornTimestamp=1550714389360, bornHost=/10.13.83.7:58033, storeTimestamp=1550714389364, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002EF5, commitLogOffset=12021, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=12, CONSUME_START_TIME=1550714389372, UNIQ_KEY=0A0D53071A0873D16E93692457700000, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]
Time [Thu Feb 21 09:59:49 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=12, sysFlag=0, bornTimestamp=1550714389369, bornHost=/10.13.83.7:58033, storeTimestamp=1550714389371, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000002FA7, commitLogOffset=12199, bodyCRC=1561245975, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=13, CONSUME_START_TIME=1550714389379, UNIQ_KEY=0A0D53071A0873D16E93692457790001, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 49], transactionId='null'}]]
Time [Thu Feb 21 09:59:49 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=13, sysFlag=0, bornTimestamp=1550714389375, bornHost=/10.13.83.7:58033, storeTimestamp=1550714389379, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000003059, commitLogOffset=12377, bodyCRC=1141369005, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=15, CONSUME_START_TIME=1550714389385, UNIQ_KEY=0A0D53071A0873D16E936924577F0002, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 50], transactionId='null'}]]
Time [Thu Feb 21 09:59:49 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=14, sysFlag=0, bornTimestamp=1550714389381, bornHost=/10.13.83.7:58033, storeTimestamp=1550714389382, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F000000000000310B, commitLogOffset=12555, bodyCRC=855693371, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=15, CONSUME_START_TIME=1550714389385, UNIQ_KEY=0A0D53071A0873D16E93692457850003, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 51], transactionId='null'}]]
Time [Thu Feb 21 09:59:49 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=15, sysFlag=0, bornTimestamp=1550714389384, bornHost=/10.13.83.7:58033, storeTimestamp=1550714389386, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000031BD, commitLogOffset=12733, bodyCRC=761548184, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest6', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=16, CONSUME_START_TIME=1550714389409, UNIQ_KEY=0A0D53071A0873D16E93692457880004, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 52], transactionId='null'}]]

可以发送所有的消息都被同一个线程处理，并且从同一个Message Queue中读取数据，可以保证数据的顺序性；

2.2局部顺序消息

生产者需要将相关业务的消息发送到同一个Message Queue，在消费端需要保证同一个Message Queue读取的消息不能被并发处理；
生产者发送消息给同一个Message Queue可以通过MessageQueueSelector来实现：

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {

        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
            return mqs.get(0);
        }
}, i);

通过在select放在中为msg指定固定的Message Queue，这里为了方便给所有的消息都指定第0个队列；
消费者保证同一个Message Queue读取的消息不能被并发处理，通过MessageListenerOrderly实现：

consumer.registerMessageListener(new MessageListenerOrderly() {
            
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                System.out.printf("Time [" + new Date().toString() + "]," +Thread.currentThread().getName() + "Receive New Messages :" + msgs + "%n");
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();

简单测试一下，分开看一下生产者和消费者日志：

Time [Thu Feb 21 10:45:43 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307463C73D16E93694E5D8B0000, offsetMsgId=0A0D530700002A9F00000000000035E9, messageQueue=MessageQueue [topic=TopicTest7, brokerName=broker-a, queueId=0], queueOffset=5]
Time [Thu Feb 21 10:45:43 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307463C73D16E93694E5D940001, offsetMsgId=0A0D530700002A9F000000000000369B, messageQueue=MessageQueue [topic=TopicTest7, brokerName=broker-a, queueId=0], queueOffset=6]
Time [Thu Feb 21 10:45:43 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307463C73D16E93694E5D980002, offsetMsgId=0A0D530700002A9F000000000000374D, messageQueue=MessageQueue [topic=TopicTest7, brokerName=broker-a, queueId=0], queueOffset=7]
Time [Thu Feb 21 10:45:43 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307463C73D16E93694E5D9D0003, offsetMsgId=0A0D530700002A9F00000000000037FF, messageQueue=MessageQueue [topic=TopicTest7, brokerName=broker-a, queueId=0], queueOffset=8]
Time [Thu Feb 21 10:45:43 CST 2019],SendResult [sendStatus=SEND_OK, msgId=0A0D5307463C73D16E93694E5D9F0004, offsetMsgId=0A0D530700002A9F00000000000038B1, messageQueue=MessageQueue [topic=TopicTest7, brokerName=broker-a, queueId=0], queueOffset=9]
Time [Thu Feb 21 10:45:56 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=5, sysFlag=0, bornTimestamp=1550717143436, bornHost=/10.13.83.7:63916, storeTimestamp=1550717143440, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000035E9, commitLogOffset=13801, bodyCRC=705268097, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest7', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=10, UNIQ_KEY=0A0D5307463C73D16E93694E5D8B0000, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 48], transactionId='null'}]]
Time [Thu Feb 21 10:45:56 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=6, sysFlag=0, bornTimestamp=1550717143444, bornHost=/10.13.83.7:63916, storeTimestamp=1550717143445, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F000000000000369B, commitLogOffset=13979, bodyCRC=1561245975, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest7', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=10, UNIQ_KEY=0A0D5307463C73D16E93694E5D940001, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 49], transactionId='null'}]]
Time [Thu Feb 21 10:45:56 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=7, sysFlag=0, bornTimestamp=1550717143448, bornHost=/10.13.83.7:63916, storeTimestamp=1550717143451, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F000000000000374D, commitLogOffset=14157, bodyCRC=1141369005, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest7', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=10, UNIQ_KEY=0A0D5307463C73D16E93694E5D980002, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 50], transactionId='null'}]]
Time [Thu Feb 21 10:45:56 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=8, sysFlag=0, bornTimestamp=1550717143453, bornHost=/10.13.83.7:63916, storeTimestamp=1550717143454, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000037FF, commitLogOffset=14335, bodyCRC=855693371, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest7', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=10, UNIQ_KEY=0A0D5307463C73D16E93694E5D9D0003, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 51], transactionId='null'}]]
Time [Thu Feb 21 10:45:56 CST 2019],ConsumeMessageThread_1Receive New Messages :[MessageExt [queueId=0, storeSize=178, queueOffset=9, sysFlag=0, bornTimestamp=1550717143455, bornHost=/10.13.83.7:63916, storeTimestamp=1550717143456, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F00000000000038B1, commitLogOffset=14513, bodyCRC=761548184, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest7', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=10, UNIQ_KEY=0A0D5307463C73D16E93694E5D9F0004, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 52], transactionId='null'}]]

3.事务消息

事务消息是指RocketMQ发送的消息和其他本地事件需要同时成功同时失败，可以理解为就是分布式事务；RocketMQ处理事务消息的大致流程如下：
1.生产者发送"待确认"消息；
2.RocketMQ接收到消息进行相关保存操作，成功以后返回状态给生产者；
3.生产者接收到的返回如果为SEND_OK状态，将执行本地事务操作；
4.根据本地事务执行的结果，生产者执行commit还是rollback；
5.如果第四步生产者执行操作失败，服务器会在经过固定时间段对状态为"待确认"的消息发起回查请求；
6.生产者接收到回查请求后根据本地事务的结果返回commit还是rollback；
7.服务器收到结果后执行相关操作。

接下来看一下官方提供的实例
TransactionProducerTest生产者类，类似DefaultMQProducer，主要设置了一个事务监听器类TransactionListener，用于开始本地事务会给服务器的回查接口；

public class TransactionProducerTest {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("transactionProducerGroupName");
        producer.setNamesrvAddr("192.168.237.128:9876");
        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 < 1; i++) {
            try {
                Message msg = new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                System.out.println("start send message " + msg);
                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();
    }
}

还有一个监听器类TransactionListenerImpl

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) {
        System.out.println("executeLocalTransaction");
        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());
        System.out.println("checkLocalTransaction:status = " + status);
        if (null != status) {
            switch (status) {
            case 0:
                return LocalTransactionState.UNKNOW;
            case 1:
                return LocalTransactionState.COMMIT_MESSAGE;
            case 2:
                return LocalTransactionState.ROLLBACK_MESSAGE;
            default:
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

以上监听器需要实现两个接口方法，分别是执行本地事务的方法和用于被服务器回调的方法；运行以上生产者相关日志如下：

start send message Message{topic='TopicTest1234', flag=0, properties={KEYS=KEY0, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48], transactionId='null'}
executeLocalTransaction
SendResult [sendStatus=SEND_OK, msgId=0A0D5307383473D16E936A31CF040000, offsetMsgId=null, messageQueue=MessageQueue [topic=TopicTest1234, brokerName=broker-a, queueId=2], queueOffset=47]
checkLocalTransaction:status = 0
checkLocalTransaction:status = 0
checkLocalTransaction:status = 0

从日志可以看出，首先发送"待确认"消息，发送返回为SEND_OK；然后执行本地事务，实例中返回的是一个LocalTransactionState.UNKNOW状态，导致服务器一直调用回查方法checkLocalTransaction，同时消费端一直没有消息被消费；做简单代码改动，将本地事务的执行结果改成LocalTransactionState.COMMIT_MESSAGE，生产者消费者日志如下：

start send message Message{topic='TopicTest1234', flag=0, properties={KEYS=KEY0, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48], transactionId='null'}
executeLocalTransaction
SendResult [sendStatus=SEND_OK, msgId=0A0D530740F073D16E936A3A2DC10000, offsetMsgId=null, messageQueue=MessageQueue [topic=TopicTest1234, brokerName=broker-a, queueId=2], queueOffset=58]
Time [Thu Feb 21 15:03:17 CST 2019],ConsumeMessageThread_3Receive New Messages :[MessageExt [queueId=2, storeSize=278, queueOffset=0, sysFlag=8, bornTimestamp=1550732597697, bornHost=/10.13.83.7:55029, storeTimestamp=1550732597758, storeHost=/10.13.83.7:10911, msgId=0A0D530700002A9F0000000000008853, commitLogOffset=34899, bodyCRC=613185359, reconsumeTimes=0, preparedTransactionOffset=34610, toString()=Message{topic='TopicTest1234', flag=0, properties={MIN_OFFSET=0, REAL_TOPIC=TopicTest1234, MAX_OFFSET=1, KEYS=KEY0, TRAN_MSG=true, UNIQ_KEY=0A0D530740F073D16E936A3A2DC10000, WAIT=true, PGROUP=transactionProducerGroupName, TAGS=TagA, REAL_QID=2}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48], transactionId='0A0D530740F073D16E936A3A2DC10000'}]]

可以看到消息状态在服务器端被修改，这样消费端就可以消费此消息；

总结

本文首先介绍了RocketMQ发送消息的通讯模式，然后重点介绍了延迟消息，顺序消息以及事务消息，并且结合实例进行分析。

示例代码地址

https://github.com/ksfzhaohui...
https://gitee.com/OutOfMemory...

rocketmq 科技新闻

Qemo

Qemo

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

消息疯狂堆积！RocketMQ 出 Bug 了?

用过 MQ 的同学，可能会遇到过消息堆积的问题。而我最近也踩上了这个坑，但是发现结果竟然是这么一个意料之外的原因而导致的。那一晚月和风高，我正准备踏上回家的路，突然收到告警短信轰炸！于是乎赶回公司赶紧打开电脑，登上 RocketMQ 后台查看

LCFlxfldy 2020-08-17

深入理解RocketMQ延迟消息

延迟消息是实际开发中一个非常有用的功能，本文第一部分从整体上介绍秒级精度延迟消息的实现思路，在第二部分结合RocketMQ的延迟消息实现，进行细致的讲解，点出关键部分的源码。第三步介绍延迟消息与消息重试的关系。Broker判断这是一个延迟消息后，将其通过临

IT农场 2020-11-13

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn 2020-07-27

(五)消息队列

此时，为了提高用户体验和吞吐量，其实可以异步地调用系统B、C、D的接口。所以，我们可以通过引入消息中间件，再系统A执行完后将数据写入到消息队列中直接返回，而BCD系统则负责监听执行。通过以上分析可知，消息中间件最佳选型为RocketMQ。

LCFlxfldy 2020-07-05

RocketMQ(2)---核心概念、特性、使用等

对于RocketMQ而言，感觉官方提供的东西还是可以的：https://github.com/apache/rocketmq/tree/master/docs/cn

lypgcs 2020-06-27

RocketMq producer 发送一条消息所经过的流程

　　RocketMq producer 在发送一条消息时候，从 producer --nameSrv -- Broker 中间经过了什么样子的数据交互。从NameSrv 中获取topic 配置的相关信息，比如 broker 地址，队列数 之类的。

陈晨软件五千言 2020-06-17

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

qingyuerji 2020-06-14

RocketMQ系列（四）顺序消费

折腾了好长时间才写这篇文章，顺序消费，看上去挺好理解的，就是消费的时候按照队列中的顺序一个一个消费；而并发消费，则是消费者同时从队列中取消息，同时消费，没有先后顺序。RocketMQ也有这两种方式的实现，但是在实践的过程中，就是不能顺序消费，好不容易能够实

MojitoBlogs 2020-06-14

RocketMQ系列（五）广播与延迟消息

今天要给大家介绍RocketMQ中的两个功能，一个是“广播”，这个功能是比较基础的，几乎所有的mq产品都是支持这个功能的；另外一个是“延迟消费”，这个应该算是RocketMQ的特色功能之一了吧。广播是把消息发送给订阅了这个主题的所有消费者。首先，消费者端的

lypgcs 2020-06-14

RocketMQ系列（一）基本概念

RocketMQ是阿里出品的一款开源的消息中间件，让其声名大噪的就是它的事务消息的功能。在企业中，消息中间件选择使用RocketMQ的还是挺多的，这一系列的文章都是针对RocketMQ的，咱们先从RocketMQ的一些基本概念和环境的搭建开始聊起。Rock

陈晨软件五千言 2020-06-14

RocketMq 在Netty 下是如何进行消息封装传输

　　要了解rocketMq 需要知道 数据在 rocketMq 中 是如何进行传输，在底层的结构到底是一个什么亚子，这个需要我们对Netty 对字符编解码有一些了解。　　我们从生产者发送消息，broker 接收消息 为例，来开展底层对消息结构。　　1. 如

meilongwhpu 2020-06-13

RocketMQ

然后执行‘start mqnamesrv.cmd’，启动NAMESERVER。成功后会弹出提示框，此框勿关闭。#set it false if you don‘t want use dashboard.default true. #set the mess

陈晨软件五千言 2020-06-11

RocketMQ集群监控平台搭建

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为

qingyuerji 2020-06-09

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

MojitoBlogs 2020-06-09

RocketMQ事务消息

RocketMQ事务消息是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。Producer 已经把消息成功发送到

meilongwhpu 2020-06-08

聊一聊高并发高可用那些事 - Kafka篇

偶尔推出一个优惠活动时，请求量极速上升。看了几个招聘网站，提到较多的消息队列有：RabbitMQ、RocketMQ、Kafka 以及 Redis 的消息队列和发布订阅模式。在高可用环境中，Kafka 需要部署多台，避免 Kafka 宕机后，服务无法访问。K

meilongwhpu 2020-06-08

RocketMQ-Namesrv源码解析

Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker部署相对复杂，Broker分为Master和Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master和Slave

lypgcs 2020-06-07

深度剖析 Kafka/RocketMQ 顺序消息的一些坑

我觉得这个问题问得很频繁，而且非常经典，在这里我就以 Kafka 为例子，说说我对 Kafka 顺序消息的一些理解吧，如有理解不对的地方麻烦留言指点一下。针对以上第二点，下面盘点一下 Kafka 集群中有哪些意外情况会打乱消息的顺序。

MojitoBlogs 2020-06-04

RocketMQ源码 — 九、 RocketMQ延时消息

上一节消息重试里面提到了重试的消息可以被延时消费，其实除此之外，用户发送的消息也可以指定延时时间，然后在指定延时时间之后投递消息，然后被consumer消费。broker收到消息在准备将消息写入存储的时候，判断是延时消息则更改Message的topic为延

meilongwhpu 2020-05-30

rocketmq初识

通常一个消息队列需要掌握的知识点有Topic（主体）、Producer（生产者）、Consumer（消费者）、Queue（队列）、Delivery Semantics. 蛋疼的是不同的消息队列关于这些名词叫法不一样，含义也不是很精确。所以阿里起了一个项目O

LCFlxfldy 2020-05-29
Qemo

Qemo

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号