spring cloud alibaba RocketMQ 最佳实践

概述

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 最佳实践 ，至此就结束了，如有疑问，欢迎评论区留言。