Spring Cloud Stream RabbitMQ 简单实现

川长思鸟来 2022-05-09 06:46 172阅读 0赞

参考官方文档:[http://cloud.spring.io/spring-cloud-static/Finchley.SR1/single/spring-cloud.html\#\_rabbitmq\_binder][http_cloud.spring.io_spring-cloud-static_Finchley.SR1_single_spring-cloud.html_rabbitmq_binder]

环境：win10 rabbit 默认配置

# 一、添加依赖 #

<dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>

# 二、配置文件 #

spring:
      application:
        name: rabbitmq-client
      cloud:
        stream:
          bindings:
            chen2018:  #通道名
              destination: chen-test #目的地
      rabbitmq:
        host: localhost
        port: 5672

# 三、构建消息发送者 #

##   1.先创建通道 ##

public interface ChenService {
      @Output("chen2018")
      MessageChannel chen();
    
    }

##   2.扫描接口 ##

@EnableBinding(ChenService.class)

##   3.发送消息类 ##

为了防止中文乱码加了消息头

@RestController
    public class TestController {
    
      @Autowired
      private RabbitTemplate rabbitTemplate;
    
      @Autowired
      private ChenService chenService;
    
      @GetMapping("/send")
      public Boolean sendMsg(@RequestParam String msg){
        //获取MessageChanel
        MessageChannel messageChannel = chenService.chen();
        Map<String,Object> headers = new HashMap<>();
        headers.put("charset-encoding","UTF-8");
        headers.put("content-type",MediaType.TEXT_PLAIN_VALUE);
        return messageChannel.send(new GenericMessage<>(msg,headers));
      }
    }

# 四、构建消息订阅者 #

##   1.构建订阅通道 ##

public interface ChenService {
      @Input("chen2018")
      SubscribableChannel chen();
    }

##   2.扫描接口 ##

@EnableBinding(ChenService.class)

##   3.多种编程模型根据实际情况选择 ##

@Autowired
    	private ChenService chenService;
    
    	@PostConstruct
    	public void init(){//接口编程
    		//获取Subscribe chanel
    		SubscribableChannel subscribableChannel = chenService.chen();
    		subscribableChannel.subscribe(message -> {
    			MessageHeaders headers = message.getHeaders();
    			String encoding = (String) headers.get("charset-encoding");
    			String text = (String)headers.get("content-type");
    			byte[] content = (byte[]) message.getPayload();
    			try {
    				System.out.println("接收到消息："+ new String(content,encoding));
    			} catch (UnsupportedEncodingException e) {
    				e.printStackTrace();
    			}
    		});
    	}
    	@StreamListener("chen2018")
    	public void onMessage(byte[] data){//注解编程
    		System.out.println("onMessage(byte[]):"+data);
    	}
    
    	@StreamListener("chen2018")
    	public void onMessage(String data){//注解编程
    		System.out.println("onMessage(String):"+data);
    	}
    
    	@ServiceActivator(inputChannel = "chen2018") //Spring Integration注解
    	public void onServiceActivator(String data){
    		System.out.println("onServiceActivator(String):"+data);
    	}

###     建议:使用同一种编程模型       相同的编程模型重复执行，不同的编程模型轮流执行 ###

# 五、测试结果 #

访问：[http://localhost:8088/send?msg=加油][http_localhost_8088_send_msg]！fighting

![70][]

[http_cloud.spring.io_spring-cloud-static_Finchley.SR1_single_spring-cloud.html_rabbitmq_binder]: http://cloud.spring.io/spring-cloud-static/Finchley.SR1/single/spring-cloud.html#_rabbitmq_binder
[http_localhost_8088_send_msg]: http://localhost:8088/send?msg=%E5%8A%A0%E6%B2%B9%EF%BC%81fighting
[70]: /images/20220509/f5d11a7447b84f81a28be08e3abfd80c.png