RocketMQ 生产者核心配置和核心知识

时间：2020-02-24 01:08:22 阅读：95 评论：0 收藏：0 [点我收藏+]

一、RocketMQ 4.X 生产者常见核心配置

compressMsgBodyOverHowmuch ：消息超过默认字节4096后进行压缩
retryTimesWhenSendFailed : 失败重发次数
maxMessageSize : 最大消息配置，默认128k
topicQueueNums : 主题下面的队列数量，默认是4
autoCreateTopicEnable : 是否自动创建主题 Topic, 开发建议为 true，生产要为 false
defaultTopicQueueNums : 自动创建服务器不存在的 Topic，默认创建的队列数
autoCreateSubscriptionGroup: 是否允许 Broker 自动创建订阅组，建议线下开发开启，线上关闭
brokerClusterName : 集群名称
brokerId : 0表示Master主节点，大于0表示从节点
brokerIP1 : Broker 服务地址
brokerRole : Broker 角色 ASYNC_MASTER/ SYNC_MASTER/ SLAVE
deleteWhen : 每天执行删除过期文件的时间，默认每天凌晨4点
flushDiskType ：刷盘策略, 默认为 ASYNC_FLUSH (异步刷盘)，另外是 SYNC_FLUSH (同步刷盘)
listenPort ： Broker 监听的端口号
mapedFileSizeCommitLog ：单个 conmmitlog 文件大小，默认是 1GB
mapedFileSizeConsumeQueue：ConsumeQueue 每个文件默认存30W条，可以根据项目调整
storePathRootDir : 存储消息以及一些配置信息的根目录默认为用户的 ${HOME}/store
storePathCommitLog：commitlog 存储目录默认为 ${storePathRootDir}/commitlog
storePathIndex：消息索引存储路径
syncFlushTimeout : 同步刷盘超时时间
diskMaxUsedSpaceRatio ：检测可用的磁盘空间大小，超过后会写入报错

二、Broker 消息投递状态有四种

FLUSH_DISK_TIMEOUT ：没有在规定时间内完成刷盘 (刷盘策略需要为 SYNC_FLUSH 才会出这个错误)
FLUSH_SLAVE_TIMEOUT ：主从模式下，Broker 是 SYNC_MASTER，没有在规定时间内完成主从同步
SLAVE_NOT_AVAILABLE ：主从模式下，Broker 是 SYNC_MASTER，没有找到被配置成 Slave 的 Broker
SEND_OK ：发送成功，没有发生上面的三种问题

三、RocketMQ 消息生产和消费异常重试和阈值设定

生产者 Producer 重试（异步和 SendOneWay下配置无效）

消息重投（保证数据的高可靠性），本身内部支持重试，默认次数是2，如果网络情况比较差，或者跨集群则建改多几次

消费端重试

原因：消息处理异常、broker 端到 consumer 端各种问题，如网络原因闪断，消费处理失败，ACK 返回失败等等问题。
注意：
- 重试间隔时间配置 ,默认每条消息最多重试 16 次
- messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 超过重试次数人工补偿
- 消费端去重
- 一条消息无论重试多少次，这些重试消息的 Message ID，key 不会改变。
- 消费重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息

技术分享图片

PayController 类代码如下：

package net.xdclass.xdclassmq.controller;

import net.xdclass.xdclassmq.jms.JmsConfig;
import net.xdclass.xdclassmq.jms.PayProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;


    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        Message message = new Message(JmsConfig.TOPIC,"taga" ,"6688" , ("hello xdclass rocketmq = "+text).getBytes() );

        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);

        return new HashMap<>();
    }

}

PayController 类

JmsConfig 类代码如下：

package net.xdclass.xdclassmq.jms;

public class JmsConfig {
    public static final String NAME_SERVER = "192.168.159.129:9876;192.168.159.130:9876";

    public static final String TOPIC = "xdclass_pay_test_topic_666";

}

JmsConfig 类

PayProducer 类代码如下：

package net.xdclass.xdclassmq.jms;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.stereotype.Component;

@Component
public class PayProducer {

    private String producerGroup = "pay_producer_group";

    private DefaultMQProducer producer;

    public  PayProducer(){
        producer = new DefaultMQProducer(producerGroup);

        //生产者投递消息重试次数
        producer.setRetryTimesWhenSendFailed(3);

      //指定NameServer地址，多个地址以 ; 隔开
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);

        start();
    }

    public DefaultMQProducer getProducer(){
        return this.producer;
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown(){
        this.producer.shutdown();
    }
}

PayConsumer 类代码如下：

package net.xdclass.xdclassmq.jms;

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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class PayConsumer {


    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "pay_consumer_group";

    public  PayConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(JmsConfig.TOPIC, "*");


        consumer.registerMessageListener( new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                int times = msg.getReconsumeTimes();
                System.out.println("重试次数="+times);

                try {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

                String topic = msg.getTopic();
                String body = new String(msg.getBody(), "utf-8");
                String tags = msg.getTags();
                String keys = msg.getKeys();

                // 模拟报错
                if(keys.equalsIgnoreCase("6688")){
                    throw new Exception();
                }

                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {

                System.out.println("消费异常");
                //如果重试2次不成功，则记录，人工介入
                if(times >= 2){
                    System.out.println("重试次数大于2，记录数据库，发短信通知开发人员或者运营人员");
                    //TODO 记录数据库，发短信通知开发人员或者运营人员
                    //告诉broker，消息成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            }
        });

        consumer.start();
        System.out.println("consumer start ...");
    }

}

四、RocketMQ 异步发送消息

官方文档：https://rocketmq.apache.org/docs/simple-example/

@RequestMapping("/api/v1/pay_cb")
public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

　　Message message = new Message(JmsConfig.TOPIC,"taga" ,"66881" , ("hello xdclass rocketmq = "+text).getBytes() );

　　payProducer.getProducer().send(message, new SendCallback() {
　　　　@Override
　　　　public void onSuccess(SendResult sendResult) {
　　　　　　System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());
　　　　}

　　　　@Override
　　　　public void onException(Throwable e) {
　　　　　　e.printStackTrace();
　　　　　　//补偿机制，根据业务情况进行使用，看是否进行重试
　　　　}
　　});

　　return new HashMap<>();
}

注意：官方例子：如果异步发送消息，调用 producer.shutdown() 后会失败。异步发送：不会重试，发送总次数等于1。

五、RocketMQ OneWay 发送消息及多种场景对比

SYNC：应用场景：重要通知邮件、报名短信通知、营销短信系统等
ASYNC ：应用场景：对RT时间敏感，可以支持更高的并发，回调成功触发相对应的业务，比如注册成功后通知积分系统发放优惠券
ONEWAY：一次发送，无需要等待响应。使用场景：主要是日志收集，适用于某些耗时非常短，但对可靠性要求并不高的场景，也就是 LogServer，只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答

@RequestMapping("/api/v1/pay_cb")
public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
    Message message = new Message(JmsConfig.TOPIC,"taga" ,"66881" , ("hello xdclass rocketmq = "+text).getBytes() );
    payProducer.getProducer().sendOneway(message);
    return new HashMap<>();
}

汇总对比

发送方式	发送 TPS	发送结果反馈	可靠性
同步发送	快	有	不丢失
异步发送	快	有	不丢失
单向发送	最快	无	可能丢失

六、RocketMQ 延迟消息

什么是延迟消息？

Producer 将消息发送到消息队列 RocketMQ 服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到 Consumer 进行消费，该消息即延迟消息，目前支持固定延迟精度的消息。

固定精度为："1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"

技术分享图片

使用 message.setDelayTimeLevel(xxx); xxx是级别，1表示配置里面的第一个级别，2表示第二个级别。

@RequestMapping("/api/v1/pay_cb")
public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
    Message message = new Message(JmsConfig.TOPIC,"taga" ,"66881" , ("hello xdclass rocketmq = "+text).getBytes() );
    message.setDelayTimeLevel(2);// 表示5s后进行投递
    payProducer.getProducer().sendOneway(message);
    return new HashMap<>();
}

使用场景

通过消息触发一些定时任务，比如在某一固定时间点向用户发送提醒消息。
消息生产和消费有时间窗口要求：比如在天猫电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在 30 分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。如支付未完成，则关闭订单。如已完成支付则忽略。

RocketMQ 还有定时消息功能，目前开源版本还不支持，商业版本则有，两者使用场景类似。

七、RocketMQ 生产消息使用 MessageQueueSelector 投递到 Topic 下指定的 queue

应用场景：顺序消息，分摊负载。

默认 Topic 下的 queue 数量是4，可以配置。

import net.xdclass.xdclassmq.jms.JmsConfig;
import net.xdclass.xdclassmq.jms.PayProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;


    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        Message message = new Message(JmsConfig.TOPIC,"taga" ,"66881" , ("hello xdclass rocketmq = "+text).getBytes() );

        //同步发送
//      SendResult sendResult =  payProducer.getProducer().send(message, new MessageQueueSelector() {
//           @Override
//            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
//               int queueNum = Integer.parseInt(arg.toString());
//               return mqs.get(queueNum);
//            }
//
//      }, 3);
//      System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());

        //异步发送到指定queue， SendCallback不能用lambda表达式，有两个函数需要被实现
        payProducer.getProducer().send(message, (mqs, msg, arg) -> {
            int queueNum = Integer.parseInt(arg.toString());
            return mqs.get(queueNum);

        }, 0, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());
            }

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

        return new HashMap<>();
    }
}

支持同步，异步发送指定的 MessageQueue。

注意：选择的 queue 数量必须小于配置的，否则会出错。

八、顺序消息在电商和证券系统中的应用场景

什么是顺序消息：消息的生产和消费顺序一致
- 全局顺序：topic下面全部消息都要有序(少用)
  - 性能要求不高，所有的消息严格按照 FIFO 原则进行消息发布和消费的场景，并行度成为消息系统的瓶颈, 吞吐量不够.
  - 在证券处理中，以人民币兑换美元为例子，在价格相同的情况下，先出价者优先处理，则可以通过全局顺序的方式按照 FIFO 的方式进行发布和消费
- 局部顺序：只要保证一组消息被顺序消费即可（RocketMQ 使用）
  - 性能要求高
  - 电商的订单创建，同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息、订单交易成功消息都会按照先后顺序来发布和消费（阿里巴巴集团内部电商系统均使用局部顺序消息，既保证业务的顺序，同时又能保证业务的高性能）
顺序发布：对于指定的一个 Topic，客户端将按照一定的先后顺序发送消息
顺序消费：对于指定的一个 Topic，按照一定的先后顺序接收消息，即先发送的消息一定会先被客户端接收到。
注意：
- 顺序消息暂不支持广播模式
- 顺序消息不支持异步发送方式，否则将无法严格保证顺序

九、顺序消息的使用

生产端保证发送消息有序，且发送到同一个 Topic 的同个 queue 里面，RocketMQ 的确是能保证 FIFO 的。

例子：订单的顺序流程是：创建、付款、物流、完成，订单号相同的消息会被先后发送到同一个队列中，根据 MessageQueueSelector 里面自定义策略，根据同个业务 id 放置到同个 queue 里面，如订单号取模运算再放到 selector 中，同一个模的值都会投递到同一条 queue。

public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
    //如果是订单号是字符串，则进行hash，得到一个hash值
    Long id = (Long) arg;
    long index = id % mqs.size();
    return mqs.get((int)index);
}

消费端要在保证消费同个 Topic 里的同个队列，不应该用 MessageListenerConcurrently，应该使用 MessageListenerOrderly，自带单线程消费消息，不能再 Consumer 端再使用多线程去消费，消费端分配到的 queue 数量是固定的，集群消会锁住当前正在消费的队列集合的消息，所以会保证顺序消费。

下面代码来演示生产者投递消息和消费者消费消息，项目目录如下：

技术分享图片

ProductOrder 类

package net.xdclass.xdclassmq.domain;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

public class ProductOrder implements Serializable {

    //订单id
    private long orderId;

    //操作类型
    private String type;


    public long getOrderId() {
        return orderId;
    }

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

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public ProductOrder() {
    }

    public ProductOrder(long orderId, String type) {
        this.orderId = orderId;
        this.type = type;

    }


    public static List<ProductOrder> getOrderList() {

        List<ProductOrder> list = new ArrayList<>();
        list.add(new ProductOrder(111L, "创建订单"));
        list.add(new ProductOrder(222L, "创建订单"));
        list.add(new ProductOrder(111L, "支付订单"));
        list.add(new ProductOrder(222L, "支付订单"));
        list.add(new ProductOrder(111L, "完成订单"));
        list.add(new ProductOrder(333L, "创建订单"));
        list.add(new ProductOrder(222L, "完成订单"));
        list.add(new ProductOrder(333L, "支付订单"));
        list.add(new ProductOrder(333L, "完成订单"));

        return list;

    }

    @Override
    public String toString() {
        return "ProductOrder{" +
                "orderId=" + orderId +
                ", type=‘" + type + ‘\‘‘ +
                ‘}‘;
    }
}

ProductOrder 类

PayController 类

package net.xdclass.xdclassmq.controller;

import net.xdclass.xdclassmq.domain.ProductOrder;
import net.xdclass.xdclassmq.jms.JmsConfig;
import net.xdclass.xdclassmq.jms.PayProducer;
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 org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    @RequestMapping("/api/v2/pay_cb")
    public Object callback() throws Exception {

        List<ProductOrder> list = ProductOrder.getOrderList();

        for (int i = 0; i < list.size(); i++) {
            ProductOrder order = list.get(i);
            Message message = new Message(JmsConfig.ORDERLY_TOPIC, "",
                    order.getOrderId() + "", order.toString().getBytes());

            SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long id = (Long) arg;
                    long index = id % mqs.size();
                    return mqs.get((int) index);
                }
            }, order.getOrderId());


            System.out.printf("发送结果=%s, sendResult=%s ,orderid=%s, type=%s\n", sendResult.getSendStatus(), sendResult.toString(), order.getOrderId(), order.getType());

        }
        return new HashMap<>();
    }
}

JmsConfig 类

package net.xdclass.xdclassmq.jms;

public class JmsConfig {
    public static final String NAME_SERVER = "192.168.159.129:9876;192.168.159.130:9876";

    public static final String TOPIC = "xdclass_pay_test_topic_888";

    public static final String ORDERLY_TOPIC = "xdclass_pay_order_topic_orderly";

}

PayConsumer 类

package net.xdclass.xdclassmq.jms;

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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class PayConsumer {


    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "pay_consumer_group";

    public  PayConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(JmsConfig.TOPIC, "*");

        consumer.registerMessageListener( new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                int times = msg.getReconsumeTimes();
                System.out.println("重试次数="+times);

                try {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

                String topic = msg.getTopic();
                String body = new String(msg.getBody(), "utf-8");
                String tags = msg.getTags();
                String keys = msg.getKeys();

                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            } catch (Exception e) {
                System.out.println("消费异常");
                //如果重试2次不成功，则记录，人工介入
                if(times >= 2){
                    System.out.println("重试次数大于2，记录数据库，发短信通知开发人员或者运营人员");
                    //TODO 记录数据库，发短信通知开发人员或者运营人员
                    //告诉broker，消息成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            }
        });

        consumer.start();
        System.out.println("consumer start ...");
    }

}

PayConsumer 类

PayOrderlyConsumer 类

package net.xdclass.xdclassmq.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class PayOrderlyConsumer {


    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "pay_orderly_consumer_group";

    public PayOrderlyConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(JmsConfig.ORDERLY_TOPIC, "*");


        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                MessageExt msg = msgs.get(0);
                try {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));

                    //做业务逻辑操作 TODO

                    return ConsumeOrderlyStatus.SUCCESS;

                } catch (Exception e) {

                    e.printStackTrace();
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }

        });

        consumer.start();
        System.out.println("consumer start ...");
    }

}

PayProducer 类

package net.xdclass.xdclassmq.jms;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.stereotype.Component;

@Component
public class PayProducer {

    private String producerGroup = "pay_producer_group";

    private DefaultMQProducer producer;


    public  PayProducer(){
        producer = new DefaultMQProducer(producerGroup);

        //生产者投递消息重试次数
        producer.setRetryTimesWhenSendFailed(3);

        //指定NameServer地址，多个地址以 ; 隔开
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);

        start();
    }

    public DefaultMQProducer getProducer(){
        return this.producer;
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start(){
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown(){
        this.producer.shutdown();
    }


}

PayProducer 类

RocketMQ 生产者核心配置和核心知识

原文：https://www.cnblogs.com/jwen1994/p/12353909.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)