springcloudstream 3.X整合_spring cloud stream 整合pulsar

springcloudstream 3.X整合

最近开始整和springcloudstream 刚开始整合2.X 还能使用@Input、@Output 到了自己手写的时候就用了3.X 发现网上教程很少官网目前也是看不懂对新手还是蛮不友好 所以决定发一篇博客

1、构建微服务

2、父pom导入依赖

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.5</version>
    </parent>
    
    <dependencyManagement>

        <dependencies>

            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>2020.0.4</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

        </dependencies>
        
    </dependencyManagement>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

3、client pom导入依赖

微服务架构 这边使用的eureka充当服务注册 eureka就不多赘述

    <dependencies>

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
		<!-- 主要依赖 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

    </dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

3.1、yml配置

这边注意 如果是输出管道 以 xxx-out-index 输入管道 xxx-in-index 详情请看官网

spring:
  application:
    name: stream-client8001
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: 
                port: 5672
                username: 
                password: 
      bindings:
        msg-out-0: 
          destination: lyc-message 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

3.2 开始编码

这边就是接口实现一下

@Service
public class MqServiceImpl implements MqService {
	
	3.X以后导入这个 
	@Resource
	private StreamBridge streamBridge;

	@Override
	public void sendMessage(String message) {
		// send方法中第一个参数是bindingName 也就是 msg-out-0  个人习惯问题 写成了常量
		streamBridge.send(MyMessageChannel.MESSAGE_OUTPUT, message);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

3.2.1 写接口 发请求

@RestController
public class MqController {

	private final MqService mqService;

	@GetMapping(value = "/{message}")
	public String sendMessage(@PathVariable("message") String message){
		mqService.sendMessage(message);
		return "ok";
	}

	public MqController(MqService mqService) {
		this.mqService = mqService;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.3、发送方搞定后就开始写接收方

pom和发送方是相同的 yml有一点区别 前面那个是发送方 所以是out 这边需要接收 所以是in

spring:
  application:
    name: stream-client8002
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: 
                port: 
                username: 
                password: 
      bindings:
        msg-in-0:
          destination: lyc-message
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

@Component
public class MyListener {

	// 方法名字  和 msg-in-0 前缀一致
	@Bean
	public Consumer<String> msg() {
		return System.out::println;
	}
}
1
2
3
4
5
6
7
8
9

到这里就可以接收到消息了 到这里只是 能初步实现发送和接收 可以初步参考
如果想要点到点 和订阅 可以尝试加入group

        msg-out-0:
          destination: lyc-message
          group
1
2
3