SpringCloud终结篇之消息驱动stream大集合

创建子工程  stream-sample

编写pom文件



org.springframework.boot
spring-boot-starter-web


org.springframework.boot
spring-boot-starter-actuator


org.springframework.cloud
spring-cloud-starter-stream-rabbit



创建启动引导类  StreamApplication

&＃064;SpringBootApplication
public class StreamApplication {
public static void main(String[] args) {
SpringApplication.run(StreamApplication.class, args);
}
}

创建配置文件

spring.application.name&＃061;stream-sample
server.port&＃061;63003
# RabbitMQ连接字符串
spring.rabbitmq.host&＃061;192.168.0.201
spring.rabbitmq.port&＃061;5672
spring.rabbitmq.username&＃061;guest
spring.rabbitmq.password&＃061;guest
# 绑定Channel到broadcast
spring.cloud.stream.bindings.myTopic-consumer.destination&＃061;broadcast
spring.cloud.stream.bindings.myTopic-producer.destination&＃061;broadcast
# 消息分组示例
spring.cloud.stream.bindings.group-consumer.destination&＃061;group-topic
spring.cloud.stream.bindings.group-producer.destination&＃061;group-topic
spring.cloud.stream.bindings.group-consumer.group&＃061;Group-A
## 消息分区配置
## 打开消费者的消费分区功能
spring.cloud.stream.bindings.group-consumer.consumer.partitioned&＃061;true
## 两个消息分区
spring.cloud.stream.bindings.group-producer.producer.partition-count&＃061;2
# SpEL (Key resolver) 可以定义复杂表达式生成Key
# 我们这里用最简化的配置&＃xff0c;只有索引参数为1的节点&＃xff08;消费者&＃xff09;&＃xff0c;才能消费消息
spring.cloud.stream.bindings.group-producer.producer.partition-key-expression&＃061;1
# 当前消费者实例总数
spring.cloud.stream.instanceCount&＃061;2
# 最大值instanceCount-1&＃xff0c;当前实例的索引号
spring.cloud.stream.instanceIndex&＃061;1
# 延迟消息配置
spring.cloud.stream.bindings.delayed-consumer.destination&＃061;delayed-topic
spring.cloud.stream.bindings.delayed-producer.destination&＃061;delayed-topic
spring.cloud.stream.rabbit.bindings.delayed-producer.producer.delayed-exchange&＃061;true
# 异常消息&＃xff08;单机版重试&＃xff09;
spring.cloud.stream.bindings.error-consumer.destination&＃061;error-out-topic
spring.cloud.stream.bindings.error-producer.destination&＃061;error-out-topic
# 重试次数&＃xff08;本机重试&＃xff09;
# 次数&＃061;1相当于不重试
spring.cloud.stream.bindings.error-consumer.consumer.max-attempts&＃061;2
# 异常消息&＃xff08;requeue重试&＃xff09;
spring.cloud.stream.bindings.requeue-consumer.destination&＃061;requeue-topic
spring.cloud.stream.bindings.requeue-producer.destination&＃061;requeue-topic
# 必须把max-attempts设置为1&＃xff0c;否则requeue不能生效
spring.cloud.stream.bindings.requeue-consumer.consumer.max-attempts&＃061;1
spring.cloud.stream.bindings.requeue-consumer.group&＃061;requeue-group
# 仅对当前requeue-consumer&＃xff0c;开启requeue
spring.cloud.stream.rabbit.bindings.requeue-consumer.consumer.requeueRejected&＃061;true
# 默认全局开启requeue
# spring.rabbitmq.listener.default-requeue-rejected&＃061;true
# 死信队列配置
spring.cloud.stream.bindings.dlq-consumer.destination&＃061;dlq-topic
spring.cloud.stream.bindings.dlq-producer.destination&＃061;dlq-topic
spring.cloud.stream.bindings.dlq-consumer.consumer.max-attempts&＃061;2
spring.cloud.stream.bindings.dlq-consumer.group&＃061;dlq-group
# 开启死信队列&＃xff08;默认 topic.dlq&＃xff09;
spring.cloud.stream.rabbit.bindings.dlq-consumer.consumer.auto-bind-dlq&＃061;true
# Fallback配置
spring.cloud.stream.bindings.fallback-consumer.destination&＃061;fallback-topic
spring.cloud.stream.bindings.fallback-producer.destination&＃061;fallback-topic
spring.cloud.stream.bindings.fallback-consumer.consumer.max-attempts&＃061;2
spring.cloud.stream.bindings.fallback-consumer.group&＃061;fallback-group
# input channel -> fallback-topic.fallback-group.errors
management.security.enabled&＃061;false
management.endpoints.web.exposure.include&＃061;*
management.endpoint.health.show-details&＃061;always

创建 Topic

延时消息

public interface DelayedTopic {
String INPUT &＃061; "delayed-consumer";
String OUTPUT &＃061; "delayed-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

死信队列

public interface DlqTopic {
String INPUT &＃061; "dlq-consumer";
String OUTPUT &＃061; "dlq-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

异常消息

public interface ErrorTopic {
String INPUT &＃061; "error-consumer";
String OUTPUT &＃061; "error-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

fallback降级

public interface FallbackTopic {
String INPUT &＃061; "fallback-consumer";
String OUTPUT &＃061; "fallback-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

分区分组

public interface GroupTopic {
String INPUT &＃061; "group-consumer";
String OUTPUT &＃061; "group-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

重入队列

public interface RequeueTopic {
String INPUT &＃061; "requeue-consumer";
String OUTPUT &＃061; "requeue-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

我的 消息

public interface MyTopic {
String INPUT &＃061; "myTopic-consumer";
String OUTPUT &＃061; "myTopic-producer";
&＃064;Input(INPUT)
SubscribableChannel input();
&＃064;Output(OUTPUT)
MessageChannel output();
}

创建stream 流  消息消费者

&＃064;Slf4j
&＃064;EnableBinding(value &＃061; {
Sink.class,
MyTopic.class,
GroupTopic.class,
DelayedTopic.class,
ErrorTopic.class,
RequeueTopic.class,
DlqTopic.class,
FallbackTopic.class
}
)
public class StreamConsumer {
private AtomicInteger count &＃061; new AtomicInteger(1);
&＃064;StreamListener(Sink.INPUT)
public void consume(Object payload) {
log.info("message consumed successfully, payload&＃061;{}", payload);
}
// 自定义消息广播
&＃064;StreamListener(MyTopic.INPUT)
public void consumeMyMessage(Object payload) {
log.info("My message consumed successfully, payload&＃061;{}", payload);
}
// 消息分组 & 消费分区示例
&＃064;StreamListener(GroupTopic.INPUT)
public void consumeGroupMessage(Object payload) {
log.info("Group message consumed successfully, payload&＃061;{}", payload);
}
// 延迟消息示例
&＃064;StreamListener(DelayedTopic.INPUT)
public void consumeDelayedMessage(MessageBean bean) {
log.info("Delayed message consumed successfully, payload&＃061;{}", bean.getPayload());
}
// 异常重试&＃xff08;单机版&＃xff09;
&＃064;StreamListener(ErrorTopic.INPUT)
public void consumeErrorMessage(MessageBean bean) {
log.info("Are you OK?");
if (count.incrementAndGet() % 3 &＃061;&＃061; 0) {
log.info("Fine, thank you. And you?");
count.set(0);
} else {
log.info("What&＃039;s your problem?");
throw new RuntimeException("I&＃039;m not OK");
}
}
// 异常重试&＃xff08;联机版-重新入列&＃xff09;
&＃064;StreamListener(RequeueTopic.INPUT)
public void requeueErrorMessage(MessageBean bean) {
log.info("Are you OK?");
try {
Thread.sleep(3000L);
} catch (Exception e) {
}
// throw new RuntimeException("I&＃039;m not OK");
}
// 死信队列
&＃064;StreamListener(DlqTopic.INPUT)
public void consumeDlqMessage(MessageBean bean) {
log.info("Dlq - Are you OK?");
if (count.incrementAndGet() % 3 &＃061;&＃061; 0) {
log.info("Dlq - Fine, thank you. And you?");
} else {
log.info("Dlq - What&＃039;s your problem?");
throw new RuntimeException("I&＃039;m not OK");
}
}
// Fallback &＃043; 升级版本
&＃064;StreamListener(FallbackTopic.INPUT)
public void goodbyeBadGuy(MessageBean bean,
&＃064;Header("version") String version) {
log.info("Fallback - Are you OK?");
if ("1.0".equalsIgnoreCase(version)) {
log.info("Fallback - Fine, thank you. And you?");
} else if ("2.0".equalsIgnoreCase(version)) {
log.info("unsupported version");
throw new RuntimeException("I&＃039;m not OK");
} else {
log.info("Fallback - version&＃061;{}", version);
}
}
// 降级流程
&＃064;ServiceActivator(inputChannel &＃061; "fallback-topic.fallback-group.errors")
public void fallback(Message message) {
log.info("fallback entered");
}
}


创建一个 messageBean

&＃064;Data
public class MessageBean {
private String payload;
}

最后一步 创建 Controller  

&＃064;RestController
&＃064;Slf4j
public class Controller {
&＃064;Autowired
private MyTopic producer;
&＃064;Autowired
private GroupTopic groupTopicProducer;
&＃064;Autowired
private DelayedTopic delayedTopicProducer;
&＃064;Autowired
private ErrorTopic errorTopicProducer;
&＃064;Autowired
private RequeueTopic requeueTopicProducer;
&＃064;Autowired
private DlqTopic dlqTopicProducer;
&＃064;Autowired
private FallbackTopic fallbackTopicProducer;
// 简单广播消息
&＃064;PostMapping("send")
public void sendMessage(&＃064;RequestParam(value &＃061; "body") String body) {
producer.output().send(MessageBuilder.withPayload(body).build());
}
// 消息分组和消息分区
&＃064;PostMapping("sendToGroup")
public void sendMessageToGroup(&＃064;RequestParam(value &＃061; "body") String body) {
groupTopicProducer.output().send(MessageBuilder.withPayload(body).build());
}
// 延迟消息
&＃064;PostMapping("sendDM")
public void sendDelayedMessage(
&＃064;RequestParam(value &＃061; "body") String body,
&＃064;RequestParam(value &＃061; "seconds") Integer seconds) {
MessageBean msg &＃061; new MessageBean();
msg.setPayload(body);
log.info("ready to send delayed message");
delayedTopicProducer.output().send(
MessageBuilder.withPayload(msg)
.setHeader("x-delay", seconds * 1000)
.build());
}
// 异常重试&＃xff08;单机版&＃xff09;
&＃064;PostMapping("sendError")
public void sendErrorMessage(&＃064;RequestParam(value &＃061; "body") String body) {
MessageBean msg &＃061; new MessageBean();
msg.setPayload(body);
errorTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// 异常重试&＃xff08;联机版 - 重新入列&＃xff09;
&＃064;PostMapping("requeue")
public void sendErrorMessageToMQ(&＃064;RequestParam(value &＃061; "body") String body) {
MessageBean msg &＃061; new MessageBean();
msg.setPayload(body);
requeueTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// 死信队列测试
&＃064;PostMapping("dlq")
public void sendMessageToDlq(&＃064;RequestParam(value &＃061; "body") String body) {
MessageBean msg &＃061; new MessageBean();
msg.setPayload(body);
dlqTopicProducer.output().send(MessageBuilder.withPayload(msg).build());
}
// fallback &＃043; 升版
&＃064;PostMapping("fallback")
public void sendMessageToFallback(
&＃064;RequestParam(value &＃061; "body") String body,
&＃064;RequestParam(value &＃061; "version", defaultValue &＃061; "1.0") String version) {
MessageBean msg &＃061; new MessageBean();
msg.setPayload(body);
fallbackTopicProducer.output().send(
MessageBuilder.withPayload(msg)
.setHeader("version", version)
.build());
}
}

附&＃xff1a;

1.    下载插件
https://www.rabbitmq.com/community-plugins.html

找到rabbitmq_delayed_message_exchange
下载对应版本的插件&＃xff0c;3.6和3.7版本插件不一样

2. 下载以后解压&＃xff0c;copy到rabbitmq安装目录下的plugins文件夹

3.    安装插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

4.    安装完一定要重启RabbitMQ&＃xff0c;不是单单重启UI管理界面&＃xff01;
如果只是单单调用rabbitmqctl  stop_app然后再rabbitmqctl  start_app是没有作用的&＃xff01;
正确的步骤是先rabbitmqctl stop&＃xff0c;然后再直接执行rabbitmq-server

如果以上步骤还能使延迟队列生效&＃xff0c;在重启完之后&＃xff0c;换一个新的topic名字就好了
 

本文地址：https://blog.csdn.net/weixin_38305866/article/details/109941990