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

红太狼 2023-02-28 04:20 44阅读 0赞

# 写在前面 #

这里介绍下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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70]

基本配置

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 的相关转换，实现

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70 1]  
同步发送，代码示例

/** * 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

<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;
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70]: /images/20230209/a52fff5eb1164bca90be42714b1510f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70 1]: /images/20230209/187b8724d530495a9ab0e273021fddab.png