springboot如何盈利_面试官：小伙子，你给我简单说一下RocketMQ 整合 Spring Boot吧...

客官°小女子只卖身不卖艺 2023-01-10 03:49 122阅读 0赞

前言

在使用SpringBoot的starter集成包时，要特别注意版本。因为SpringBoot集成RocketMQ的starter依赖是由Spring社区提供的，目前正在快速迭代的过程当中，不同版本之间的差距非常大，甚至基础的底层对象都会经常有改动。例如如果使用rocketmq-spring-boot-starter:2.0.4版本开发的代码，升级到目前最新的rocketmq-spring-boot-starter:2.1.1后，基本就用不了了

应用结构

![60d6969085702cec46b76e3bbffed1ee.png][]

TestController: 测试入口, 有基本消息测试和事务消息测试

TopicListener: 是监听"topic"这个主题的普通消息监听器

TopicTransactionListener: 是监听"topic"这个主题的事务消息监听器, 和TopicTransactionRocketMQTemplate绑定(一一对应关系)

Customer: 是测试消息体的一个entity对象

TopicTransactionRocketMQTemplate: 是扩展自RocketMQTemplate的另一个RocketMQTemplate, 专门用来处理某一个业务流程, 和TopicTransactionListener绑定(一一对应关系)

pom.xml

org.apache.rocketmq:rocketmq-spring-boot-starter:2.1.1, 引用的springboot版本是2.0.5.RELEASE<?xml version="1.0" encoding="UTF-8"?>

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

com.mrathena.middle.ware

rocket.mq.springboot

1.0.0

org.springframework.boot

spring-boot-dependencies

2.4.0

pom

import

org.projectlombok

lombok

1.18.12

org.slf4j

slf4j-api

1.7.30

ch.qos.logback

logback-classic

1.2.3

org.apache.rocketmq

rocketmq-spring-boot-starter

2.1.1

org.springframework.boot

spring-boot-starter

org.springframework

spring-core

org.springframework

spring-webmvc

org.springframework

spring-aop

org.springframework

spring-context

org.springframework

spring-messaging

com.fasterxml.jackson.core

jackson-databind

org.springframework.boot

spring-boot-starter-web

org.springframework.boot

spring-boot-starter-test

org.springframework

spring-messaging

com.fasterxml.jackson.core

jackson-databind

io.springfox

springfox-swagger-ui

2.9.2

io.springfox

springfox-swagger2

2.9.2

org.apache.maven.plugins

maven-compiler-plugin

3.8.1

1.8

UTF-8

application.ymlserver:

servlet:

context-path:

port: 80

rocketmq:

name-server: 116.62.162.48:9876

producer:

group: producer

Customerpackage com.mrathena.rocket.mq.entity;

import lombok.AllArgsConstructor;

import lombok.Getter;

import lombok.NoArgsConstructor;

import lombok.Setter;

@Getter

@Setter

@NoArgsConstructor

@AllArgsConstructor

public class Customer \{

private String username;

private String nickname;

生产者 TestControllerpackage com.mrathena.rocket.mq.controller;

import com.mrathena.rocket.mq.configuration.TopicTransactionRocketMQTemplate;

import com.mrathena.rocket.mq.entity.Customer;

import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.client.producer.SendCallback;

import org.apache.rocketmq.client.producer.SendResult;

import org.apache.rocketmq.spring.core.RocketMQTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.Message;

import org.springframework.messaging.MessageHeaders;

import org.springframework.messaging.core.MessagePostProcessor;

import org.springframework.messaging.support.MessageBuilder;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;

import java.util.Map;

@Slf4j

@RestController

@RequestMapping("test")

public class TestController \{

private static final String TOPIC = "topic";

@Autowired

private RocketMQTemplate rocketMQTemplate;

@Autowired

private TopicTransactionRocketMQTemplate topicTransactionRocketMQTemplate;

@GetMapping("base")

public Object base() \{

// destination: topic/topic:tag, topic或者是topic拼接tag的整合体

// payload: 荷载即消息体

// message: org.springframework.messaging.Message, 是Spring自己封装的类, 和RocketMQ的Message不是一个类, 里面没有tags/keys等内容

rocketMQTemplate.send(TOPIC, MessageBuilder.withPayload("你好").setHeader("你是谁", "你猜").build());

// tags null

rocketMQTemplate.convertAndSend(TOPIC, "tag null");

// tags empty, 证明 tag 要么有值要么null, 不存在 empty 的 tag

rocketMQTemplate.convertAndSend(TOPIC + ":", "tag empty ?");

// 只有 tag 没有 key

rocketMQTemplate.convertAndSend(TOPIC + ":a", "tag a");

rocketMQTemplate.convertAndSend(TOPIC + ":b", "tag b");

// 有 property, 即 RocketMQ 基础 API 里面, Message(String topic, String tags, String keys, byte\[\] body) 里面的 key

// rocketmq-spring-boot-starter 把 userProperty 和其他的一些属性都糅合在 headers 里面可, 具体可以参考 org.apache.rocketmq.spring.support.RocketMQUtil.addUserProperties

// 获取某个自定义的属性的时候, 直接 headers.get("自定义属性key") 就可以了

Map properties = new HashMap<>();

properties.put("property", 1);

properties.put("another-property", "你好");

rocketMQTemplate.convertAndSend(TOPIC, "property 1", properties);

rocketMQTemplate.convertAndSend(TOPIC + ":a", "tag a property 1", properties);

rocketMQTemplate.convertAndSend(TOPIC + ":b", "tag b property 1", properties);

properties.put("property", 5);

rocketMQTemplate.convertAndSend(TOPIC, "property 5", properties);

rocketMQTemplate.convertAndSend(TOPIC + ":a", "tag a property 5", properties);

rocketMQTemplate.convertAndSend(TOPIC + ":c", "tag c property 5", properties);

// 消息后置处理器, 可以在发送前对消息体和headers再做一波操作

rocketMQTemplate.convertAndSend(TOPIC, "消息后置处理器", new MessagePostProcessor() \{

/\*\*

\* org.springframework.messaging.Message

\*/

@Override

public Message> postProcessMessage(Message> message) \{

Object payload = message.getPayload();

MessageHeaders messageHeaders = message.getHeaders();

return message;

\});

// convertAndSend 底层其实也是 syncSend

// syncSend

log.info("\{\}", rocketMQTemplate.syncSend(TOPIC, "sync send"));

// asyncSend

rocketMQTemplate.asyncSend(TOPIC, "async send", new SendCallback() \{

@Override

public void onSuccess(SendResult sendResult) \{

log.info("onSuccess");

@Override

public void onException(Throwable e) \{

log.info("onException");

\});

// sendOneWay

rocketMQTemplate.sendOneWay(TOPIC, "send one way");

// 这个我还是不太清楚是干嘛的? 跑的时候会报错!!!

// Object receive = rocketMQTemplate.sendAndReceive(TOPIC, "你好", String.class);

// log.info("\{\}", receive);

return "success";

@GetMapping("transaction")

public Object transaction() \{

Message message = MessageBuilder.withPayload(new Customer("mrathena", "你是谁")).build();

// 这里使用的是通过 @ExtRocketMQTemplateConfiguration(group = "anotherProducer") 扩展出来的另一个 RocketMQTemplate

log.info("\{\}", topicTransactionRocketMQTemplate.sendMessageInTransaction(TOPIC, message, null));

log.info("\{\}", topicTransactionRocketMQTemplate.sendMessageInTransaction(TOPIC + ":tag-a", message, null));

return "success";

配置 TopicTransactionRocketMQTemplatepackage com.mrathena.rocket.mq.configuration;

import org.apache.rocketmq.spring.annotation.ExtRocketMQTemplateConfiguration;

import org.apache.rocketmq.spring.core.RocketMQTemplate;

/\*\*

\* 一个事务流程和一个RocketMQTemplate需要一一对应

\* 可以通过 @ExtRocketMQTemplateConfiguration(注意该注解有@Component注解) 来扩展多个 RocketMQTemplate

\* 注意: 不同事务流程的RocketMQTemplate的producerGroup不能相同

\* 因为MQBroker会反向调用同一个producerGroup下的某个checkLocalTransactionState方法, 不同流程使用相同的producerGroup的话, 方法可能会调用错

\*/

@ExtRocketMQTemplateConfiguration(group = "anotherProducer")

public class TopicTransactionRocketMQTemplate extends RocketMQTemplate \{\}

消费者 TopicListenerpackage com.mrathena.rocket.mq.listener;

import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.spring.annotation.ConsumeMode;

import org.apache.rocketmq.spring.annotation.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;

import org.springframework.stereotype.Component;

/\*\*

\* 最简单的消费者例子

\* topic: 主题

\* consumerGroup: 消费者组

\* selectorType: 过滤方式, TAG:标签过滤,仅支持标签, SQL92:SQL过滤,支持标签和属性

\* selectorExpression: 过滤表达式, 根据selectorType定, TAG时, 写标签如 "a || b", SQL92时, 写SQL表达式

\* consumeMode: CONCURRENTLY:并发消费, ORDERLY:顺序消费

\* messageModel: CLUSTERING:集群竞争消费, BROADCASTING:广播消费

\*/

@Slf4j

@Component

@RocketMQMessageListener(topic = "topic",

// 只过滤tag, 不管headers中的key和value

// selectorType = SelectorType.TAG,

// 必须指定selectorExpression, 可以过滤tag和headers中的key和value

// selectorType = SelectorType.SQL92,

// 不限tag

// selectorExpression = "\*",

// 不限tag, 和 \* 一致

// selectorExpression = "",

// 只要tag为a的消息

// selectorExpression = "a",

// 要tag为a或b的消息

// selectorExpression = "a || b",

// SelectorType.SQL92时, 可以跳过tag, 直接用headers里面的key和value来判断

// selectorExpression = "property = 1",

// tag不为null

// selectorExpression = "TAGS is not null",

// tag为empty, 证明tag不会是empty, 要么有值要么null

// selectorExpression = "TAGS = ''",

// SelectorType.SQL92时, 即过滤tag, 又过滤headers里面的key和value

// selectorExpression = "(TAGS is not null and TAGS = 'a') and (property is not null and property between 4 and 6)",

// 并发消费

consumeMode = ConsumeMode.CONCURRENTLY,

// 顺序消费

// consumeMode = ConsumeMode.ORDERLY,

// 集群消费

messageModel = MessageModel.CLUSTERING,

// 广播消费

// messageModel = MessageModel.BROADCASTING,

consumerGroup = "consumer"

)

public class TopicListener implements RocketMQListener \{

public void onMessage(String s) \{

log.info("\{\}", s);

消费者 TopicTransactionListenerpackage com.mrathena.rocket.mq.listener;

import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;

import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;

import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;

import org.apache.rocketmq.spring.support.RocketMQHeaders;

import org.springframework.messaging.Message;

import org.springframework.messaging.MessageHeaders;

import org.springframework.stereotype.Component;

@Slf4j

@Component

@RocketMQTransactionListener(rocketMQTemplateBeanName = "topicTransactionRocketMQTemplate")

public class TopicTransactionListener implements RocketMQLocalTransactionListener \{

@Override

public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) \{

// message: org.springframework.messaging.Message, 是Spring自己封装的类, 和RocketMQ的Message不是一个类, 里面没有tags/keys等内容

// 一般来说, 并不会在这里处理tags/keys等内容, 而是根据消息体中的某些字段做不同的操作, 第二个参数也可以用来传递一些数据到这里

log.info("executeLocalTransaction message:\{\}, object:\{\}", message, o);

log.info("payload: \{\}", new String((byte\[\]) message.getPayload()));

MessageHeaders headers = message.getHeaders();

log.info("tags: \{\}", headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS));

log.info("rocketmq\_TOPIC: \{\}", headers.get("rocketmq\_TOPIC"));

log.info("rocketmq\_QUEUE\_ID: \{\}", headers.get("rocketmq\_QUEUE\_ID"));

log.info("rocketmq\_MESSAGE\_ID: \{\}", headers.get("rocketmq\_MESSAGE\_ID"));

log.info("rocketmq\_TRANSACTION\_ID: \{\}", headers.get("rocketmq\_TRANSACTION\_ID"));

log.info("TRANSACTION\_CHECK\_TIMES: \{\}", headers.get("TRANSACTION\_CHECK\_TIMES"));

log.info("id: \{\}", headers.get("id"));

return RocketMQLocalTransactionState.UNKNOWN;

@Override

public RocketMQLocalTransactionState checkLocalTransaction(Message message) \{

log.info("checkLocalTransaction message:\{\}", message);

// 在调用了checkLocalTransaction后, 另一个常规消息监听器才能收到消息

return RocketMQLocalTransactionState.COMMIT;

最后

欢迎关注公众号：前程有光，领取一线大厂Java面试题总结+各知识点学习思维导+一份300页pdf文档的Java核心知识点总结！

[60d6969085702cec46b76e3bbffed1ee.png]: /images/20221119/e82817cc6d924702a259b8a078327ec7.png

springboot如何盈利_面试官：小伙子，你给我简单说一下RocketMQ 整合 Spring Boot吧...

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关面试官：简单说一下使用Kafka的场景吧

相关面试官：MySQL事务的底层原理说一下吧

相关面试官：来给我说一下Spring MVC的执行过程？

相关 springboot如何盈利_面试官：小伙子，你给我简单说一下RocketMQ 整合 Spring Boot吧...

相关面试官：给我说一下 Spring MVC 拦截器的原理？

相关面试官：来给我说一下Spring MVC的执行过程？

相关面试官：说一下你常用的加密算法

相关面试：你说一下 MyBatis 事务吧！

相关面试官：说一下你常用的加密算法

相关面试官：说一说 Spring Boot 自动配置原理吧，我懵逼了...