系列文章导航: Spring Cloud Alibaba微服务解决方案
常用MQ产品的选择
目前主流的MQ产品有kafka、RabbitMQ、ActiveMQ、RocketMQ等。在MQ选型时可以参照这篇文章选择合适的MQ产品。
RocketMQ及控制台搭建
RocketMQ的搭建可以参考这篇文章。
RocketMQ控制台的搭建可以参考这篇文章。
RocketMQ与Spring Boot整合
pom.xml中添加如下依赖
org.apache.rocketmq
rocketmq-spring-boot-starter
2.0.3
配置文件中添加如下配置
rocketmq:
name-server: 172.17.0.102:9876
producer:
# 构建rocketMQtemplate必须指定group
group: test-producer
生产者发送消息
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ShareServiceImpl implements IShareService {
private final @NonNull RocketMQTemplate rocketMQTemplate;
@Override
public Share aduitById(Integer id, ShareAuditDTO shareAuditDTO) {
//第一个参数为主题topic，第二个参数为消息体对象
rocketMQTemplate.convertAndSend("add-bonus",
UserAddBonusMsgDTO.builder()
.userId(share.getUserId())
.bonus(50).build());
return share;
}
}
发送后可在RocketMQ-Console控制台查看
消费者监听消息并进行业务处理
@Service
//在@RocketMQMessageListener注解中设置消费者group和监听主题topic
@RocketMQMessageListener(consumerGroup = "test-consumer", topic = "add-bonus")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
//将监听的消息体对象指定为RocketMQListener类的泛型
public class AddBonusListener implements RocketMQListener{
@Override
public void onMessage(UserAddBonusMsgDTO userAddBonusMsgDTO) {
//处理业务逻辑
// do something ...
}
}
事务消息
事务消息会将发送的消息进行标记，在收到commit指令后才会进行消息投递。事务消息的执行流程如下图：
这个方案有个前提，它假设消费者总是有能力成功处理消息。如果消费者消费失败，可以进行重试，如果依然失败，会进入死信队列。进入死信队列的消息可以重新入队，或者人工介入去处理。当然，也可以对消费失败的消息加入补偿机制，来保证数据的一致性。
发送半消息
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ShareServiceImpl implements IShareService {
private final @NonNull RocketMQTemplate rocketMQTemplate;
@Override
public String auditById(Integer id, ShareAuditDTO shareAuditDTO, UserAddBonusMsgDTO userAddBonusMsgDTO) {
String transactionId = UUID.randomUUID().toString();
rocketMQTemplate.sendMessageInTransaction(
//group
"tx-add-bonus-group",
//topic
"add-bonus",
//消息体
MessageBuilder.withPayload(userAddBonusMsgDTO)
//设置header,执行本地事务时可以获取使用
.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
.setHeader("share_id", id).build()
,
//设置arg,执行本地事务时可以获取使用
shareAuditDTO
);
return "success";
}
}
本地事务执行及状态检查
//指定监听本地事务的group
@RocketMQTransactionListener(txProducerGroup = "tx-add-bonus-group")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class AddbonusTransactionListener implements RocketMQLocalTransactionListener {
private final @NonNull IShareService shareService;
private final @NonNull RocketmqTransactionLogMapper rocketmqTransactionLogMapper;
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
//执行本地事务
MessageHeaders headers = message.getHeaders();
String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
Integer shareId = Integer.valueOf((String) headers.get("share_id"));
try {
//service方法中将生成的transactionId进行存储
shareService.auditByIdInDBWithRocketMQLog(shareId, (ShareAuditDTO) o, transactionId);
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
return RocketMQLocalTransactionState.ROLLBACK;
}
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
//检查本地事务是否执行成功
String transactionId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
//通过检查本地事务日志记录，确认本地事务是否执行成功
RocketmqTransactionLog rocketmqTransactionLog = rocketmqTransactionLogMapper.selectOne(
RocketmqTransactionLog.builder()
.transactionId(transactionId).build()
);
if (rocketmqTransactionLog != null) {
return RocketMQLocalTransactionState.COMMIT;
}
return RocketMQLocalTransactionState.ROLLBACK;
}
}
RocketMQ与Spring Cloud Stream整合
添加依赖
com.alibaba.cloud
spring-cloud-starter-stream-rocketmq
yml添加配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: ${rocketmq.name-server}
bindings:
output:
destination: stream-test-topic
#名称需和@output注解中指定的名称一致
my-output:
#指定输出的topic
destination: stream-my-topic
#名称需和@input注解中指定的名称一致
my-input:
#指定接收的topic
destination: stream-my-topic
#如果整合RocketMQ，必须设置group
#如果整合其他MQ，可留空
group: binder-my-group
生产者
编写自定义Source接口，使用@Output注解指定消息管道的名称，需与yml配置文件中bindings下配置的管道名称一致，才能在IOC注入时获取到yml中指定的destination值做为发送的topic，如果不对应会默认使用@Output注解指定的值作为topic，自定义消费者Sink时同理。
public interface MySource {
String MY_OUTPUT = "my-output";
@Output(MY_OUTPUT)
MessageChannel output();
}
启动类注册自定义Source接口
@SpringBootApplication
@EnableBinding({Source.class, MySource.class})
public class ContentCenterApplication {
public static void main(String[] args) {
SpringApplication.run(ContentCenterApplication.class, args);
}
}
调用自定义Source接口发送消息
@RestController
@RequestMapping
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class TestController {
private final @NonNull MySource mySource;
@GetMapping("/test-stream")
public String testStream() {
mySource.output().send(MessageBuilder.withPayload("测试stream-2消息体").build());
return "success";
}
}
消费者
编写自定义Sink接口，使用@Input注解指定消息管道的名称，需与yml配置文件中bindings下配置的管道名称一致。
public interface MySink {
String MY_INPUT = "my-input";
@Input(MY_INPUT)
SubscribableChannel input();
}
启动类注册自定义Sink接口
@SpringBootApplication
@EnableBinding({Sink.class, MySink.class})
public class UserCenterApplication {
public static void main(String[] args) {
SpringApplication.run(UserCenterApplication.class, args);
}
}
使用@StreamListener注解指定自定义Sink监听消息并处理
@Service
@Slf4j
public class MyTestStreamConsumer {
@StreamListener(MySink.MY_INPUT)
public void receive(String messageBody) {
log.info("自定义接收器，通过stream收到消息：messageBody = {}", messageBody);
}
}
使用Source发送事务消息
使用Spring Cloud Stream发送RocketMQ的事务消息时，Source接口发送的消息无法在方法调用时指定事务消息的监听group，需在yml配置中进行设置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 172.17.0.102:9876
bindings:
#管道名称需与stream.bindings对应
output:
producer:
#标注为事务消息
transactional: true
#事务消息监听group名称，对应@RocketMQTransactionListener注解txProducerGroup属性
group: tx-add-bonus-group
bindings:
output:
destination: add-bonus
发送事务消息时，仅能够通过headers发送参数。
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ShareServiceImpl implements IShareService {
private final @NonNull Source source;
@Override
public Share auditById(Integer id, ShareAuditDTO shareAuditDTO) {
//....
String transactionId = UUID.randomUUID().toString();
source.output().send(
MessageBuilder.withPayload(
UserAddBonusMsgDTO.builder()
.userId(share.getUserId())
.bonus(50).build())
//设置header,执行本地事务时可以获取使用
.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
.setHeader("share_id", id)
.setHeader("dto", JSON.toJSONString(shareAuditDTO))
.build()
);
//....
return share;
}
}
Spring Cloud Stream消息过滤消费
参考这篇文章。
Spring Cloud Stream异常处理
参考这篇文章。
Spring Cloud Stream概念及注解
参考这篇文章。
