springboot集成RocketMQ,三种方式(原生Jar,springboot封装starter,阿里云Ons接入)

写在前面

这里介绍下Springboot 集成RocketMQ的三种方式

一、原生 jar（rocketmq-client）

1.1、producer

1.1.1、三个基本使用

producerGroup，定义生产者组
DefaultMQProducer，定义生产者配置
TransactionMQProducer，定义支持事务生产者

1.1.2、三种基本发送方式：

同步发送
异步发送
单项发送

同步发送，代码示例

/** * 同步发送实体对象消息 * 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式； * 特点：速度快；有结果反馈；数据可靠； * 应用场景：应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等； * * @param topic * @param tags * @param body * @return * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */
    public String syncSend(String topic, String tags, String body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg = new Message(topic /* Topic */,
                tags /* Tag */,
                ("Hello RocketMQ ").getBytes() /* Message body */
        );
        // 发送消息到一个Broker
        SendResult sendResult = producer.send(msg);
        // 通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);
        TimeUnit.SECONDS.sleep(1);
        return "{\"MsgId\":\"" + sendResult.getMsgId() + "\"}";
    }

异步发送，代码示例

/** * 异步发送消息 * 可靠异步发送：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式； * 特点：速度快；有结果反馈；数据可靠； * 应用场景：异步发送一般用于链路耗时较长,对 rt响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等； * * @param topic * @param tags * @param body * @return * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */
    public void asyncSend(String topic, String tags, String body) throws Exception {
        Message msg = new Message(topic /* Topic */,
                tags /* Tag */,
                ("Hello RocketMQ ").getBytes() /* Message body */
        );
        // 发送消息到一个Broker
        producer.send(msg, new SendCallback() {
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送结果 ： " + sendResult);
            }

            public void onException(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
        });
        TimeUnit.SECONDS.sleep(1);
    }

单项发送，代码示例

/** * 单向发送 * 单向发送：只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答；此方式发送消息的过程耗时非常短，一般在微秒级别； * 特点：速度最快，耗时非常短，毫秒级别；无结果反馈；数据不可靠，可能会丢失； * 应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集； * * @param topic * @param tags * @param body * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */
    public void oneway(String topic, String tags, String body) throws Exception {
        Message msg = new Message(topic /* Topic */,
                tags /* Tag */,
                ("Hello RocketMQ ").getBytes() /* Message body */
        );
        producer.sendOneway(msg);
        TimeUnit.SECONDS.sleep(1);
    }

1.1.3、其他发送特性

消息延迟
设置消息属性，用于消费过滤
消息队列选择器Selector
事务监听

消息延迟

/** * 延迟 消费 */
    public void delayTestListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delayGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe("delayPushMsg", "push");

            // 程序第一次启动从消息队列头获取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            consumer.setConsumeMessageBatchMaxSize(1);
            //在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                // 会把不同的消息分别放置到不同的队列中
                for (MessageExt msg : msgs) {
                    log.info("Receive message:msgId={},msgBody={},delay={} ms",
                            msg.getMsgId(),
                            new String(msg.getBody()),
                            (System.currentTimeMillis() - msg.getStoreTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

设置消息属性，用于消费过滤

/** * todo 注意这里 需要启动 broker 前，设置 支持SQL92 Filter = enable * sql filter * * @param topic * @param tags * @param body * @param i * @throws Exception */
    public void filtersql(String topic, String tags, String body, int i) throws Exception {
        //消息
        Message message = new Message(topic, tags, body.getBytes());
        //设置消息属性
        message.putUserProperty("i", String.valueOf(i));
        //发送消息
        SendResult sendresult = producer.send(message);
        System.out.println("消息结果 ：" + sendresult);
        TimeUnit.SECONDS.sleep(1);
    }

消息队列选择器Selector

/** * Order 测试 * * @param topic * @param tags * @param body * @param order * @throws Exception */
    public void orderPush(String topic, String[] tags, String body, Boolean order) throws Exception {
        // 订单列表
        List<OrderStep> orderList = this.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++) {
            if (order) {
                log.info("有序的消费，根据队列id，分配分组，启动相应的唯一消费线程");
                // 加个时间前缀
                String body1 = dateStr + body + orderList.get(i);
                Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, body1.getBytes());

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    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(),
                        body1));
            } else {
                log.info("无序的的消费，需等所有消息发布完成，在分配，根据队列id，启动相应的唯一消费线程");
                // 加个时间前缀
                String body1 = dateStr + body + orderList.get(i);
                Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, body1.getBytes());
                SendResult sendResult = producer.send(msg);

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

    }

事务监听

/** * 事务测试 * * @param topic * @param tags * @param body * @throws Exception */
    public void tasnsaction(String topic, String[] tags, String body) throws Exception {

        //创建事务监听器
        TransactionListener listener = new TransactionListener() {
            /** * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction. * * @param message Half(prepare) message * @param o Custom business parameter * @return Transaction state */
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if ("Tag1".equals(message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if ("Tag2".equals(message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else return LocalTransactionState.UNKNOW;
            }

            /** * When no response to prepare(half) message. broker will send check message to check the transaction status, and this * method will be invoked to get local transaction status. * * @param messageExt Check message * @return Transaction state */
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println(messageExt.getTags() + "消息回查！");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        };
        //set事务监听器
        transProducer.setTransactionListener(listener);
        //发送消息
        for (int i = 0; i < 3; i++) {
            Message message = new Message(topic, tags[i], "KEY" + i, (body + i).getBytes());
            SendResult sendResult = transProducer.sendMessageInTransaction(message, null);
            System.out.println("发送结果 ：" + sendResult);
            TimeUnit.SECONDS.sleep(2);
        }
    }

1.2、consumer

1.2.1、基本使用

DefaultMQPushConsumer

1.2.2、监听示例

基本配置

/** * RocketMq配置监听信息 */
    public void messageListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SpringBootRocketMqGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe("PushTopic", "push");

            // 程序第一次启动从消息队列头获取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //负载均衡模式消费
           // consumer.setMessageModel(MessageModel.BROADCASTING);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            consumer.setConsumeMessageBatchMaxSize(1);
            //在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                // 会把不同的消息分别放置到不同的队列中
                for (Message msg : msgs) {
                    System.out.println("接收到了消息：" + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

filter tag

/** * filter tag 监听 */
    public void filterTagListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tagFilterGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            //订阅的topic与tag
            consumer.subscribe("topic1", "tag1 || tag2");
            //注册消息监听器
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt msg : list) {
                        log.info("收到消息:Keys->{},body->{}", msg.getKeys(), new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

条件过滤

/** * sql92 条件过滤 */
    public void sqlFilterTagListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sqlFilterGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            //设置订阅条件
            consumer.subscribe("topic2", MessageSelector.bySql("i > 5"));
            //注册监听器
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt msg : list) {
                        log.info("收到消息:Keys->{},body->{},i ->{}",
                                msg.getKeys(),
                                new String(msg.getBody()),
                                msg.getProperty("i"));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

二、rocketmq-spring-boot-starter

Maven 坐标

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>${rocketmq-spring-boot-starter-version}</version>
        </dependency>

基本架构，不是很复杂，将RocketMQ client Jar 中的相关连接配置，Message 对象转换成了Spring Bean

springboot集成RocketMQ,三种方式(原生Jar,springboot封装starter,阿里云Ons接入)

基本配置

rocketmq.name-server=localhost:9876
rocketmq.producer.group=boot-group1
rocketmq.producer.sendMessageTimeout=300000

2.1、producer

两个模板对象

RocketMQTemplate，内置对象
extRocketMQTemplate,需自定义的RocketMQTemplate

RocketMQTemplate，其实封装了Spring message 和 rocket-client 的相关转换，实现

springboot集成RocketMQ,三种方式(原生Jar,springboot封装starter,阿里云Ons接入)
同步发送，代码示例

/** * Send string * localhost:10001/sendString */
    @GetMapping("/sendString")
    public void sendString() {
        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
        System.out.printf("syncSend1 to topic %s sendResult=%s %n", springTopic, sendResult);

        sendResult = rocketMQTemplate.syncSend(userTopic, new User().setUserAge((byte) 18).setUserName("Kitty"));
        System.out.printf("syncSend1 to topic %s sendResult=%s %n", userTopic, sendResult);

        sendResult = rocketMQTemplate.syncSend(userTopic, MessageBuilder.withPayload(
                new User().setUserAge((byte) 21).setUserName("Lester")).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build());
        System.out.printf("syncSend1 to topic %s sendResult=%s %n", userTopic, sendResult);
    }

异步发送，代码示例

/** * 异步 * Send user-defined object * localhost:10001/send-with-user-defined */
    @GetMapping("/send-with-user-defined")
    public void userDefined() {
        rocketMQTemplate.asyncSend(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")), new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                System.out.printf("async onSucess SendResult=%s %n", var1);
            }

            @Override
            public void onException(Throwable var1) {
                System.out.printf("async onException Throwable=%s %n", var1);
            }

        });
    }

单向发送，代码示例

/** * 单向发送 */
    @GetMapping("/send-with-oneWay")
    public void sendOneWay() {
        rocketMQTemplate.sendOneWay(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")));
    }

事务监听处理

/** * Send transactional messages using rocketMQTemplate * localhost:10001/send-transactional-rocketMQTemplate */
    @GetMapping("/send-transactional-rocketMQTemplate")
    public void transactionalRocketMQTemplate() {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {

                Message msg = MessageBuilder.withPayload("rocketMQTemplate transactional message " + i).
                        setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
                SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
                        springTransTopic + ":" + tags[i % tags.length], msg, null);
                System.out.printf("------rocketMQTemplate send Transactional msg body = %s , sendResult=%s %n",
                        msg.getPayload(), sendResult.getSendStatus());

                Thread.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

// todo 这里还有很多种方式，具体操作场景、差别还有待研究，学习

2.2、consumer

这里比较简单，代码示例如下

/** * StringConsumer */
@Service
@RocketMQMessageListener(topic = "${demo.rocketmq.topic}",
        consumerGroup = "string_consumer",
        selectorExpression = "${demo.rocketmq.tag}")
public class StringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.printf("------- StringConsumer received: %s \n", message);
    }
}

/** * The consumer that replying String */
@Service
@RocketMQMessageListener(topic = "${demo.rocketmq.stringRequestTopic}",
        consumerGroup = "${demo.rocketmq.stringRequestConsumer}",
        selectorExpression = "${demo.rocketmq.tag}")
public class StringConsumerWithReplyString implements RocketMQReplyListener<String, String> {

    @Override
    public String onMessage(String message) {
        System.out.printf("------- StringConsumerWithReplyString received: %s \n", message);
        return "reply string";
    }
}

其他，需自行学习…

三、阿里云 ONS

首先，对于测试学习，挺贵的哈，但是却提供了非常可靠的消息机制
使用上也很简单，引入 SDK

<!-- https://mvnrepository.com/artifact/com.aliyun.openservices/ons-client -->
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>ons-client</artifactId>
            <version>1.8.6.Final</version>
        </dependency>

3.1、配置

rocketmq:
  producer:
  producerId: GroupId#生产者id(旧版本是生产者id，新版本是groupid)，
  msgTopic: Test #生产主题
  accessKey: XXX  #连接通道
  secretKey: XXX  #连接秘钥
  onsAddr:  #生产者ons接入域名

3.2、使用

配置类

package com.tonels.spring.boot.rocketmq.producer;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Properties;

/** * rocketmq生产者启动初始化类 * */
@Component
public class RocketmqProducerInit {

    @Value("${rocketmq.producer.producerId}")
    private String producerId;

    @Value("${rocketmq.producer.accessKey}")
    private String accessKey;

    @Value("${rocketmq.producer.secretKey}")
    private String secretKey;

    @Value("${rocketmq.producer.onsAddr}")
    private String ONSAddr;

    private static Producer producer;

    @PostConstruct
    public void init(){
        System.out.println("初始化启动生产者！");
        // producer 实例配置初始化
        Properties properties = new Properties();
        //您在控制台创建的Producer ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, producerId);
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
        //设置发送超时时间，单位毫秒
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        // 设置 TCP 接入域名(此处以公共云生产环境为例)，设置 TCP 接入域名，进入 MQ 控制台的消费者管理页面，在左侧操作栏单击获取接入点获取
        properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr);
        producer = ONSFactory.createProducer(properties);
        // 在发送消息前，初始化调用start方法来启动Producer，只需调用一次即可，当项目关闭时，自动shutdown
        producer.start();
    }

    /** * 初始化生产者 * @return */
    public Producer getProducer(){
        return producer;
    }

}

发送消息

@Autowired
private RocketmqProducerInit producer;

public boolean sendMsg(String msg) {
        Long startTime = System.currentTimeMillis();
        Message message = new Message(msgTopic, tag, msg.getBytes());
        SendResult sendResult = producer.getProducer().send(message);
        if (sendResult != null) {
            System.out.println(new Date() + " Send mq message success. Topic is:" + message.getTopic() + " msgId is: " + sendResult.getMessageId());
        } else {
            logger.warn(".sendResult is null.........");
        }
        Long endTime = System.currentTimeMillis();
        System.out.println("单次生产耗时："+(endTime-startTime)/1000);
        return true;
    }