[Stream]SpringCloudStream的消息分组配置

喜欢ヅ旅行 2021-12-23 23:53 317阅读 0赞

**搭配使用**  
logback日志配置:  
[https://blog.csdn.net/a755199443/article/details/92208902][https_blog.csdn.net_a755199443_article_details_92208902]

Eureka单机服务端配置:  
[https://blog.csdn.net/a755199443/article/details/92579936][https_blog.csdn.net_a755199443_article_details_92579936]

Eureka服务端集群配置:  
[https://blog.csdn.net/a755199443/article/details/92620273][https_blog.csdn.net_a755199443_article_details_92620273]

linux部署Eureka的服务端集群:  
[https://blog.csdn.net/a755199443/article/details/92637977][https_blog.csdn.net_a755199443_article_details_92637977]

消息驱动的消息发送者配置:  
[https://blog.csdn.net/a755199443/article/details/93716471][https_blog.csdn.net_a755199443_article_details_93716471]

消息驱动的消息接收者配置:  
[https://blog.csdn.net/a755199443/article/details/93717596][https_blog.csdn.net_a755199443_article_details_93717596]

消息分组可以实现消息接收的集群.  
本博客基于**消息驱动的消息发送者配置/消息接收者配置**的基础上修改而成.

### SpringCloudStream的消息分组配置 ###

消息发送全局配置文件修改

消息发送接口修改

消息接收全局配置文件修改

消息接收接口修改

消息接收接口的实现类修改

### 消息发送全局配置文件修改 ###

增加了最后一行

spring.application.name=stream_sender
    server.port=9050
    
    #设置服务注册中心地址，指向另一个注册中心
    eureka.client.serviceUrl.defaultZone=http://user:123456@eureka1:8761/eureka/,http://user:123456@eureka2:8761/eureka/
    
    #rebbitmq 链接信息
    spring.rabbitmq.host=192.168.110.129
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456
    spring.rabbitmq.virtualHost=/
    
    # 键名为outputProduct,队列名是exchangeProduct 
    spring.cloud.stream.bindings.outputProduct.destination=exchangeProduct

### 消息发送接口修改 ###

使用了配置文件中自己设置的键名.

public interface ISendeService { 
    
    	String OUTPUT = "outputProduct";
    	@Output(OUTPUT)
    	SubscribableChannel send();
    }

### 消息接收全局配置文件修改 ###

spring.application.name=stream_reciver
    server.port=9051
    
    #设置服务注册中心地址，指向另一个注册中心
    eureka.client.serviceUrl.defaultZone=http://user:123456@eureka1:8761/eureka/,http://user:123456@eureka2:8761/eureka/
    
    #rebbitmq 链接信息
    spring.rabbitmq.host=192.168.110.129
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456
    spring.rabbitmq.virtualHost=/
    
    # 对应 MQ 是 exchange 
    spring.cloud.stream.bindings.inputProduct.destination=exchangeProduct
    
    # 具体分组 对应 MQ 是 队列名称 并且持久化队列 
    spring.cloud.stream.bindings.inputProduct.group=groupProduct

### 消息接收接口修改 ###

public interface IReceiveService { 
    
    	String INPUT = "inputProduct";
    	@Input(INPUT)
    	SubscribableChannel receive();
    }

### 消息接收接口的实现类修改 ###

@Service
    @EnableBinding({ IReceiveService.class})
    public class ReceiveService { 
    
    	@StreamListener(IReceiveService.INPUT)
    	public void onReceive(byte[] msg){ 
    		//处理消息
    		System.out.println("Receiver: "+new String(msg));
    	}
    }

[https_blog.csdn.net_a755199443_article_details_92208902]: https://blog.csdn.net/a755199443/article/details/92208902
[https_blog.csdn.net_a755199443_article_details_92579936]: https://blog.csdn.net/a755199443/article/details/92579936
[https_blog.csdn.net_a755199443_article_details_92620273]: https://blog.csdn.net/a755199443/article/details/92620273
[https_blog.csdn.net_a755199443_article_details_92637977]: https://blog.csdn.net/a755199443/article/details/92637977
[https_blog.csdn.net_a755199443_article_details_93716471]: https://blog.csdn.net/a755199443/article/details/93716471
[https_blog.csdn.net_a755199443_article_details_93717596]: https://blog.csdn.net/a755199443/article/details/93717596