SpringCloud Stream+RabbitMQ简单整合

悠悠 2022-04-18 06:27 227阅读 0赞

本篇记录Spring Cloud Stream和RabbitMQ的整合，简单实现了消息生产和消息消费。

本篇有两个项目节点互为消息的生产者和消息消费者。

# 1 父maven工程 #

## 1.1 工程结构如下： ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70]

## 1.2 pom.xml如下： ##

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      
      <groupId>com.study</groupId>
      <artifactId>cloud-ma</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      <packaging>pom</packaging>
      <name>SpringCloudStudy</name>
      <description>SpringCloudStudy</description>
      
      
      <repositories>          
    	<repository>            
    	    <id>nexus</id>            
    		<url>http://xxx.xx.xxx.xxx:8081/repository/maven-public/</url>            
    	    <releases>
    		    <enabled>true</enabled>
    	    </releases>           
    		<snapshots>
    			<enabled>true</enabled>
    		</snapshots>          
    	  </repository>               
       </repositories>
      
      <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
         <version>2.0.3.RELEASE</version>
         <relativePath/>
      </parent>
      
      <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <java.version>1.8</java.version>
         <spring-cloud.version>Finchley.RELEASE</spring-cloud.version>
      </properties>
      
      <dependencies>
      
        
    	    
    	    <dependency>
    	        <groupId>org.springframework.boot</groupId>
    	        <artifactId>spring-boot-starter-web</artifactId>
    	        <exclusions>
    	            <exclusion>
    	                <groupId>org.springframework.boot</groupId>
    	                <artifactId>spring-boot-starter-logging</artifactId>
    	            </exclusion>
    	        </exclusions>
    	    </dependency>
    	    
    	    
    	    <dependency>
    	        <groupId>org.springframework.boot</groupId>
    	        <artifactId>spring-boot-starter-log4j2</artifactId>
    	    </dependency>
    	    
    	    
    	    
    	    <dependency>
    	        <groupId>com.fasterxml.jackson.dataformat</groupId>
    	        <artifactId>jackson-dataformat-yaml</artifactId>
    	    </dependency>
    	    
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        
        
    	<dependency>
    	    <groupId>org.springframework.boot</groupId>
    	    <artifactId>spring-boot-devtools</artifactId>
    	    <optional>true</optional>
    	    <scope>true</scope>
    	</dependency>
    	
    	
    	<dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.51</version>
        </dependency>
        
        
      </dependencies>
      
      <dependencyManagement>
          <dependencies>
              <dependency>
                  <groupId>org.springframework.cloud</groupId>
                  <artifactId>spring-cloud-dependencies</artifactId>
                  <version>${spring-cloud.version}</version>
                  <type>pom</type>
                  <scope>import</scope>
              </dependency>
          </dependencies>
      </dependencyManagement>
      
      <build>
         <plugins>
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
                 <configuration>
    	                
    	                <fork>true</fork>
    	         </configuration>
             </plugin>
         </plugins>
      </build>
    
      
      <modules>
      	<module>EurekaServer</module>
      	<module>EurekaClientHi</module>
        <module>EurekaClientRibbonCustomer</module>
        <module>EurekaClientHi2</module>
        <module>EurekaClientFeignCustomer</module>
        <module>EurekaClientZuul</module>
        <module>config_server</module>
        <module>config-client</module>
        <module>config-server-svn</module>
        <module>config-client-svn</module>
        <module>StreamProvider</module>
        <module>stream-output</module>
        <module>stream-input</module>
        <module>StreamRabbitMQSelf</module>
      </modules>
    </project>

# 2 项目节点一 #

项目节点一的工程名称为stream-input-output

## 2.1 工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 1]

## 2.2 该工程的pom.xml ##

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <modelVersion>4.0.0</modelVersion>
      <parent>
        <groupId>com.study</groupId>
        <artifactId>cloud-ma</artifactId>
        <version>0.0.1-SNAPSHOT</version>
      </parent>
      
      <artifactId>stream-input</artifactId>
      <name>stream-input</name>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
      
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <scope>test</scope>
        </dependency>
      </dependencies>
    </project>

## 2.3 application.yml ##

application.yml位于src/main/resources,其配置如下：

server:
      port: 8085
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment: #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xx.xxx.xxx
                    username: xxx
                    password: xxx
                    virtual-host: / 
          bindings: 
            output:
              destination: MyOutput  #exchange名称，交换模式默认是topic
              content-type: application/json
            input: 
              destination: MyInput
              content-type: application/json

application.yml配置了 spring.cloud.stream.bindings.input.destination=MyInput、spring.cloud.stream.bindings.output.destination=MyOutput 后会在RabbitMQ 中创建一个名为MyInput交换器（exchange）和一个名为MyOutput交换器（exchange)。如果没有在RabbitMQ中创建这两个交换器，那么程序运行后会自动在RabbbitMQ中创建。

spring.cloud.stream.bindings.input.destination=MyInput是把springcloud stream的输入通道和RabbitMQ的MyInput交换器绑定，spring.cloud.stream.bindings.output.destination=MyOutput是把springcloud stream的消息输出通道和RabbitMQ的MyOutput交换器绑定。

## 2.4 Application主类 ##

package com.stream.input;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * Hello world! * */
    @SpringBootApplication
    public class StreamInputApplication { 
        public static void main( String[] args ) { 
            SpringApplication.run(StreamInputApplication.class, args);
        }
    }

## 2.5 消息生产类 ##

### 2.5.1 消息生产类–接口 ###

/** * */
    package com.stream.input.rabbitMQ.service;
    
    import org.springframework.integration.core.MessageSource;
    
    
    /** * @author mazhen * */
    public interface SendMsg { 
    	
    	public MessageSource<String>  sendTime();
    	
    }

### 2.5.2 消息生产类–实现类 ###

/** * */
    package com.stream.input.rabbitMQ.service.impl;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.context.annotation.Bean;
    import org.springframework.integration.annotation.InboundChannelAdapter;
    import org.springframework.integration.annotation.Poller;
    import org.springframework.integration.core.MessageSource;
    import org.springframework.messaging.support.GenericMessage;
    
    import com.stream.input.rabbitMQ.service.SendMsg;
    
    /** * @author mazhen * */
    
    @EnableBinding(value= { Source.class})
    public class SendMsgImpl implements SendMsg { 
        
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(SendMsgImpl.class);
    	
    	private String format="yyyy-mm-dd HH:mm:ss";
    	
        @Bean
        @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000", maxMessagesPerPoll = "1"))
        @Override
     	public MessageSource<String> sendTime() { 
        	logger.info(new SimpleDateFormat(format).format(new Date()));
        	return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
     	}
    
    }

## 2.6 消息消费类 ##

### 2.6.1 消息消费类–接口 ###

/** * */
    package com.stream.input.rabbitMQ.service;
    
    import org.springframework.messaging.Message;
    
    /** * @author mazhen * */
    public interface ReceiveMsg { 
    	public void receiveTime(Message<String> message);
    }

### 2.6.2 消息消费类–实现类 ###

/** * */
    package com.stream.input.rabbitMQ.service.impl;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Component;
    
    import com.stream.input.rabbitMQ.service.ReceiveMsg;
    
    
    /** * @author mazhen * */
    @Component
    @EnableBinding(value= { Sink.class})
    public class ReceiveMsgImpl implements ReceiveMsg { 
       
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(ReceiveMsgImpl.class);
    	
    	@Override
    	@StreamListener(Sink.INPUT)
    	public void receiveTime(Message<String> message) { 
    		logger.info("接收消息" + message.getPayload().toString());
    	}
    
    }

## 2.7 log4j2.yml ##

log4j2.yml位于src/main/resources,其配置如下：

Appenders:
        Console: #输出到控制台
          name: CONSOLE #Appender命名
          target: SYSTEM_OUT
          PatternLayout:
            pattern: "%d{yyyy-MM-dd HH:mm:ss,SSS}:%4p %t (%F:%L) - %m%n"
        RollingFile: # 输出到文件，超过256MB归档
          - name: ROLLING_FILE
            ignoreExceptions: false
    # fileName: /Users/zsq/Desktop/Project/tream-input-output/tream-input-output.log
    # filePattern: "/Users/zsq/Desktop/Project/tream-input-output/$${date:yyyy-MM} -%d{yyyy-MM-dd}-%i.log.gz"
            fileName: /stream-input-output/logs/stream-input-output.log
            filePattern: "/stream-input-output/logs/$${date:yyyy-MM}/stream-input-output-%d{yyyy-MM-dd}-%i.log.gz"
            PatternLayout:
              pattern: "%d{yyyy-MM-dd HH:mm:ss,SSS}:%4p %t (%F:%L) - %m%n"
            Policies:
              SizeBasedTriggeringPolicy:
                size: "256 MB"
            DefaultRolloverStrategy:
              max: 1000
              
    Loggers:
        Root:
          level: info
          AppenderRef:
            - ref: CONSOLE
        Logger: #单独设置某些包的输出级别
          - name: com.stream.input #复数加上-
            additivity: false #去除重复的log
            level: trace
            AppenderRef: 
              - ref: CONSOLE #复数加上-
              - ref: ROLLING_FILE #复数加上-

# 3 项目节点二 #

项目节点二的工程名称为stream-output-input

## 3.1 工程结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]

## 3.2 该工程的pom.xml ##

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <modelVersion>4.0.0</modelVersion>
      <parent>
        <groupId>com.study</groupId>
        <artifactId>cloud-ma</artifactId>
        <version>0.0.1-SNAPSHOT</version>
      </parent>
    
      <artifactId>stream-output</artifactId>
      <name>stream-output</name>
      <url>http://maven.apache.org</url>
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
      
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <scope>test</scope>
        </dependency>
      </dependencies>
    </project>

## 3.3 application.yml ##

application.yml位于src/main/resources,其配置如下：

server:
      port: 8083
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment: #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xx.xxx.xxx
                    username: xxx
                    password: xxx
                    virtual-host: / 
          bindings: 
            output:
              destination: MyInput  #exchange名称，交换模式默认是topic
              content-type: application/json
            input: 
              destination: MyOutput
              content-type: application/json

application.yml配置了 spring.cloud.stream.bindings.input.destination=MyOutput、spring.cloud.stream.bindings.output.destination=MyInput 后会在RabbitMQ 中创建一个名为MyInput交换器（exchange）和一个名为MyOutput交换器（exchange)。如果项目节点一先启动运行，那么这两个交换器已经在RabbitMQ中创建。

这样项目节点二的输入通道对应项目节点一的输出通道，项目节点二的输出通道对应项目节点一的输入通道，就可以互相作为对方的消费端和生成端。

## 3.4 Application主类 ##

package com.stream.output;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * Hello world! * */
    @SpringBootApplication
    public class StreamOutputApplication { 
        public static void main( String[] args ) { 
            SpringApplication.run(StreamOutputApplication.class, args);
        }
    }

## 3.5 消息生产类 ##

### 3.5.1 消息生产类–接口 ###

/** * */
    package com.stream.output.rabbitMQ.service;
    
    import org.springframework.integration.core.MessageSource;
    
    /** * @author mazhen * */
    public interface SendMsg { 
       public MessageSource<String> sendTime();
    }

### 3.5.2 消息生产类–实现类 ###

/** * */
    package com.stream.output.rabbitMQ.service.impl;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.context.annotation.Bean;
    import org.springframework.integration.annotation.InboundChannelAdapter;
    import org.springframework.integration.annotation.Poller;
    import org.springframework.integration.core.MessageSource;
    import org.springframework.messaging.support.GenericMessage;
    
    import com.stream.output.rabbitMQ.service.SendMsg;
    
    /** * @author mazhen * */
    @EnableBinding(Source.class)
    public class SendMsgImpl implements SendMsg { 
    	
    
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(SendMsgImpl.class);
    	
    	private String format="yyyy-mm-dd HH:mm:ss";
    
    	
        @Bean
        @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000", maxMessagesPerPoll = "1"))
        @Override
    	public MessageSource<String> sendTime() { 
       	   logger.info(new SimpleDateFormat(format).format(new Date()));
       	   return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
    	}
    
    }

## 3.6 消息消费类 ##

### 3.6.1 消息消费类–接口 ###

/** * */
    package com.stream.output.rabbitMQ.service;
    
    import org.springframework.messaging.Message;
    
    /** * @author mazhen * */
    public interface ReceiveMsg { 
    	public void receiveTime(Message<String> message);
    }

### 3.6.2 消息消费类–实现类 ###

/** * */
    package com.stream.output.rabbitMQ.service.impl;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Component;
    
    import com.stream.output.rabbitMQ.service.ReceiveMsg;
    
    /** * @author mazhen * */
    @Component
    @EnableBinding(value={ Sink.class})
    public class ReceiveMsgImpl implements ReceiveMsg { 
    
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(ReceiveMsgImpl.class);
    	
    	@Override
    	@StreamListener(Sink.INPUT)
    	public void receiveTime(Message<String> message) { 
    		logger.info("接收消息" + message.getPayload());
    	}
    
    }

## 3.7 log4j2.yml ##

Appenders:
        Console: #输出到控制台
          name: CONSOLE #Appender命名
          target: SYSTEM_OUT
          PatternLayout:
            pattern: "%d{yyyy-MM-dd HH:mm:ss,SSS}:%4p %t (%F:%L) - %m%n"
        RollingFile: # 输出到文件，超过256MB归档
          - name: ROLLING_FILE
            ignoreExceptions: false
    # fileName: /Users/zsq/Desktop/Project/tream-output-input/tream-output-intput.log
    # filePattern: "/Users/zsq/Desktop/Project/tream-output-input/$${date:yyyy-MM} -%d{yyyy-MM-dd}-%i.log.gz"
            fileName: /stream-output-input/logs/stream-output-input.log
            filePattern: "/stream-output-input/logs/$${date:yyyy-MM}/stream-output-input-%d{yyyy-MM-dd}-%i.log.gz"
            PatternLayout:
              pattern: "%d{yyyy-MM-dd HH:mm:ss,SSS}:%4p %t (%F:%L) - %m%n"
            Policies:
              SizeBasedTriggeringPolicy:
                size: "256 MB"
            DefaultRolloverStrategy:
              max: 1000
              
    Loggers:
        Root:
          level: info
          AppenderRef:
            - ref: CONSOLE
        Logger: #单独设置某些包的输出级别
          - name: com.stream.ouput #复数加上-
            additivity: false #去除重复的log
            level: trace
            AppenderRef: 
              - ref: CONSOLE #复数加上-
              - ref: ROLLING_FILE #复数加上-

# 4 测试 #

*  启动 RabbitMQ
 *  启动项目节点一
 *  启动项目节点二

在 RabbitMQ 创建了 MyInput 和 MyOutput 交换器如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]  
下面是对应的消息队列：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]  
在项目节点一和项目节点二的控制台可以看到下面的消息输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 6]

#### 欢迎关注我的微信公众号，不定期更新文章和大家一起学习共勉-![在这里插入图片描述][20191212162557284.png] ####

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70]: /images/20220418/e6a713917fca4eb09f8557c9a987f6d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 1]: /images/20220418/b97985a50e204236bd9e0d05a417e429.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]: /images/20220418/89fb50f3894f452c80eff8570a16146b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]: /images/20220418/e87225bea3154ed3967b19ed0ed7adc2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]: /images/20220418/ce1b7843abd047b6b304a4b6ff2c6cd5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]: /images/20220418/76209a6f2b5b4057b84cff0d06c3cc16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 6]: /images/20220418/e0f070278a2743bfa80b5cc637805237.png
[20191212162557284.png]: /images/20220418/fe6b66e6ad06455794aeb4b7aefdc186.png