spring cloud alibaba RocketMQ 最佳实践

发布时间:2023年12月18日

概述

github 文档地址
rocket mq example

RocketMQ 版本为 5.1.4

使用

准备工作

阅读此文需要事先准备 RocketMQ ,如有疑问,请移步 RocketMQ 服务搭建

引入依赖

此处,多依赖了一些依赖。

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bootstrap</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

创建Topic

sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic

在这里插入图片描述

代码

配置 Input 和 Output 的 Binding 信息并配合 @EnableBinding 注解使其生效

@EnableBinding({ Source.class, Sink.class })
@EnableDiscoveryClient
@SpringBootApplication
public class MqApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqApplication.class);
    }
}

配置 Binding 信息:

# 配置rocketmq的nameserver地址
spring.cloud.stream.rocketmq.binder.name-server=10.xx.xx.143:9876
# 定义name为output的binding
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=application/json
# 定义name为input的binding
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=application/json
spring.cloud.stream.bindings.input.group=test-group

发送消息代码:

public class RocketMQProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr("10.xx.xx.143:9876");
        producer.start();

        Message msg = new Message("test-topic", "tagStr", "message from rocketmq producer".getBytes());
        producer.send(msg);
    }
}

应用启动

1.增加配置,在应用的 /src/main/resources/application.properties 中添加基本配置信息

spring:
  application:
    name: mq
server:
  port: 9700

2.启动应用,支持 IDE 直接启动和编译打包后启动。

  • IDE 直接启动:找到主类 RocketMQApplication,执行 main 方法启动应用。
  • 打包编译后启动:首先执行 mvn clean package 将工程编译打包,然后执行 java -jar rocketmq-example.jar 启动应用。

消息接收

使用 name 为 output 对应的 binding 发送消息到 test-topic 这个 topic。

@Service
public class ReceiveService {

	@StreamListener("input1")
	public void receiveInput1(String receiveMsg) {
		System.out.println("input1 receive: " + receiveMsg);
	}
}

在这里插入图片描述

再扩展一个

一个服务中要使用多个 Topic ,这种情况如何解决?下面给出解决方案。

public interface Input2Sink extends Sink {
    String INPUT2 = "input2";

    @Input(Input2Sink.INPUT2)
    SubscribableChannel input2();
}
public interface Output2Source extends Source {

    String OUTPUT2 = "output2";

    @Output(Output2Source.OUTPUT2)
    MessageChannel output2();
}
@Service
public class ReceiveService {

    @StreamListener("input")
    public void receiveInput1(String receiveMsg) {
        System.out.println("input receive: " + receiveMsg);
    }

    @StreamListener("input2")
    public void receiveInput2(String receiveMsg) {
        System.out.println("input2 receive: " + receiveMsg);
    }
}
@EnableBinding(value = {Source.class, Sink.class, Output2Source.class, Input2Sink.class})
@EnableDiscoveryClient
@SpringBootApplication
public class MqApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqApplication.class);
    }
}

结果如下:
在这里插入图片描述

结束

spring cloud alibaba RocketMQ 最佳实践 ,至此就结束了,如有疑问,欢迎评论区留言。

文章来源:https://blog.csdn.net/2301_79691134/article/details/134910727
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。