【RocketMq实战第五篇】-不同类型生产者（DefaultMQProducer）

缺乏、安全感 2022-03-22 11:44 352阅读 0赞

## 前言 ##

本文来介绍RocketMQ生产者发送消息默认使用的DefaultMQProducer类。

生产者 向消息队列里写入消息，不 同的业务场景需要生产者采用不同的写入策略 。 比如同步发送、异步发送、 延迟发送、 发送事务消息等。

## 正文 ##

我们结合代码来了解一下，

代码位置在package org.apache.rocketmq.example.simple;

public class AsyncProducer {
        public static void main(
            String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException {
    
            DefaultMQProducer producer = new DefaultMQProducer("Jodie_Daily_test");
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);
    
            for (int i = 0; i < 10000000; i++) {
                try {
                    final int index = i;
                    Message msg = new Message("Jodie_topic_1023",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    producer.send(msg, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                        }
    
                        @Override
                        public void onException(Throwable e) {
                            System.out.printf("%-10d Exception %s %n", index, e);
                            e.printStackTrace();
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            producer.shutdown();
        }
    }

代码讲解

发送消息要经过几个步骤 :

（1 ）设置 Producer 的 GroupName。

（2 ）设置 lnstanceName，当一个 Jvm 需要启动多个 Producer 的时候，通过设置不同的 InstanceName来区分，不设置的话系统使用默认名称“DEFAULT”。（本例没有写）

（ 3 ）设置发送失败重试次数，当网络出现异常的时候，这个次数影响消息的重复投递次数。想保证不丢消息，可以设置多重试几次 。

（4 ）设置 NameServer 地址 。  
（5 ）组装消息并发送 。

**消息的发送有同步和异步两种方式，上面的代码使用的是异步方式 。消息发送的返回状态有如下四种 : FLUSH\_DISK\_TIMEOUT 、 FLUSH\_SLAVE\_TIMEOUT 、SLAVE\_NOT\_AVAILABLE 、SEND\_OK，不同状态在不同的刷盘策略和同步策略的配置下含义是不同的 。**

*  FLUSH\_DISK\_TIMEOUT : 表示没有在规定时间内完成刷盘(需要 Broker 的刷盘策被设置成 SYNC\_FLUSH 才会报这个错误) 。
 *  FLUSH\_SLAVE\_TIMEOUT :表示在主备方式下，并且 Broker被设 置 成 SYNC\_MASTER 方式，没有在设定时间内完成 主从同步 。
 *  SLAVE\_NOT\_AVAILABLE : 这个状态 产生的场景和 FLUSH\_SLAVE\_TIMEOUT 类似， 表示在主备 方式下，并且 Broker被设置成 SYNC\_MASTER，但是没有找到被配置成 Slave 的 Broker。
 *  SEN\_ OK :表示发送成功，发送成功的具体含义，比如消息是否已经 被存储到融盘?消息是否被同步到了 Slave上?消息在 Slave上是否被 写人磁盘?需要结合所配置的刷盘策略、主从策略来定 。 这个状态还可 以简单理解为，没有发生上面列出的 三个问题状态就是 SEND OK。

发送延迟消息

Broker收到这类消息后 ，延迟一段时间再处理， 使消息在规定的一段时间后生效。

使用方法:在创建 Message对象时，调用 setDelayTimeLevel ( int level) 方法设置延迟时间， 然后再把这个消息发送 出去。 目前延迟的时间不支 持任意设置，仅支持预设值的时间长度 ( 1s/5s/1Os/30s/Im/2m/3m/4m/5m/6m/ 7m/8m/9m/1Om/20m/30m/1h/2h)。 比如 setDelayTimeLevel(3)表示延迟 10s。

自定义消息发送规则

一个 Topic会有多个 Message Queue，如果使用 Producer的默认配置，这 个 Producer 会轮流向各个 Message Queue 发 送 消息 。 Consumer 在消费消息的 时候，会根据负载均衡策略，消费被分配到的 Message Queue，如果不经过特 定的设置，某条消息被发往哪个 Message Queue，被哪个 Consumer 消费是未 知的。

如果业务 需 要我们把消息 发 送到指定的 Message Queue 里，比如把同 一 类型 的消息都发 往 相同的 Message Queue,可以用 Message QueueSelector。

发送消息的时候，把 MessageQueueSelector 的对象作为参数，使用 public SendResult send ( Message msg, MessageQueueSelector selector, Object arg)函 数发送消 息即可 。 在 MessageQueueSelector 的实现中，根据传人的 Object参 数，或者根据 Message 消息内容确定把消息发往那个 Message Queue，返回被 选中的 Message Queue。

如果感兴趣，请加公众号

![20190128113939139.jpeg][]

[20190128113939139.jpeg]: /images/20220322/1bcd5a36250b4c4b8cc0b79c0b79bd7f.png