SpringCloud Stream+RabbitMQ自定义通道

系统管理员 2022-04-18 06:34 336阅读 0赞

SpringCloud Stream默认的消息生产通道和消费通道分别是output和input，我们也可以自定义消息生产通道和消费通道；下面对这一过程进行记录。

# 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.xxx.xxx.xxx:xxxx/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 StreamRabbitMQSelf工程 #

## 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>StreamRabbitMQSelf</artifactId>
      <name>StreamRabbitMQSelf</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>

## 2.3 自定义消息通道 ##

### 2.3.1 消息生产通道 ###

新建一个接口，这个接口中可以定义多个管道用来发送消息，可以实现向不同的exchange发送消息。

/** * */
    package com.StreamRabbitMQSelf.rabbitMQ.channels;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    /** * @author mazhen * */
    public interface SendOutputChannel { 
    	// 这里可以定义不同的通道
        String MSG_SENDER = "msgSender"; // 通道名
        
        @Output(SendOutputChannel.MSG_SENDER)
        MessageChannel msgSender();
        
    }

### 2.3.2 消息消费通道 ###

新建一个接口，这个接口中可以定义多个管道用来消费消息，可以实现从不同的exchange消费消息。

/** * */
    package com.StreamRabbitMQSelf.rabbitMQ.channels;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.SubscribableChannel;
    
    /** * @author mazhen * */
    public interface ReceiveInputChannel { 
    	// 这里可以定义不同的通道
    	String MSG_RECEIVER = "msgReceiver"; // 通道名
    
    	@Input(ReceiveInputChannel.MSG_RECEIVER)
    	SubscribableChannel msgReceiver();
    	
    }

## 2.4 application.yml ##

server:
      port: 8086
    spring:
      cloud:
        stream:
          binders: 
            defaultRabbit: 
              type: rabbit
              environment:                                      #配置rabbimq连接环境
                spring: 
                  rabbitmq:
                    host: xxx.xxx.xxx.xxx
                    username: xxx
                    password: xxx
                    virtual-host: / 
          bindings: 
            msgSender:                                          #生产者绑定，这个是消息通道的名称
              destination: message-exchange                     #exchange名称，交换模式默认是topic
              content-type: application/json
            msgReceiver:                                        #消费者绑定 这个是接受消息通道的名称
              group: for-all                                    #持久化, 也就是指定队列名称，等同于rabbitmq中的 queue, 同一个服务不同的实例，使用相同的队列名称，处理消息 
              destination: message-exchange                     #和生产者的消息交换机要相同
              content-type: application/json 
              consumer: 
                max-attempts: 3 # The number of attempts to process the message (including the first) in the event of processing failures,Default: 3
                concurrency: 1 # The concurrency setting of the consumer. Default: 1. 
          rabbit: 
            bindings: 
              msgReceiver: 
                consumer: 
                  max-concurrency: 4 # maxumum concurrency of this consumer (threads)
                  prefetch: 5 # number of prefetched messages pre consumer thread
                  requeue-rejected: false # true to requeue rejected messages, false to discard (or route to DLQ)
                  republish-to-dlq: true # republish failures to the DLQ with diagnostic headers

## 2.5 消息生产和消费类 ##

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

/** * */
    package com.StreamRabbitMQSelf.rabbitMQ.service;
    
    import org.springframework.messaging.Message;
    
    /** * @author mazhen * 消息生产类--接口 */
    public interface SendMessage { 
    	 
    	 public void sendMsgStr(String str);
    }

/** * */
    package com.StreamRabbitMQSelf.rabbitMQ.service;
    
    import org.springframework.messaging.Message;
    
    /** * @author mazhen * 消息消费类--接口 */
    public interface ReceviveMessage { 
    	public void receiveStr(Message<String> message);
    }

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

/** * */
    package com.StreamRabbitMQSelf.rabbitMQ.service.impl;
    
    import java.util.concurrent.TimeUnit;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.integration.support.MessageBuilder;
    import org.springframework.messaging.Message;
    
    import com.StreamRabbitMQSelf.rabbitMQ.channels.SendOutputChannel;
    import com.StreamRabbitMQSelf.rabbitMQ.service.SendMessage;
    
    
    /** * @author mazhen * 消息生产实现类 */
    @EnableBinding(SendOutputChannel.class)
    public class SendMessageImpl implements SendMessage { 
        
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(SendMessageImpl.class);
    	
    	@Autowired
        private SendOutputChannel sendOutputChannel;
    
    	@Override
    	public void sendMsgStr(String str) { 
    		if (!sendOutputChannel.msgSender().send(MessageBuilder.withPayload(str).build())) { 
    			 logger.error("生产者消息发送失败：" + str);
    		}
    		logger.info("[sendMsgStr]生产者消息发送:"+str);
    	}
    
    }

/** * */
    package com.StreamRabbitMQSelf.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.messaging.Message;
    
    import com.StreamRabbitMQSelf.model.CdkeyParameter;
    import com.StreamRabbitMQSelf.rabbitMQ.channels.ReceiveInputChannel;
    import com.StreamRabbitMQSelf.rabbitMQ.service.ReceviveMessage;
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    
    /** * @author mazhen * 消息消费实现类 */
    @EnableBinding(ReceiveInputChannel.class)
    public class ReceviveMessageImpl implements ReceviveMessage { 
    
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(ReceviveMessageImpl.class);
    	
    	@Override
    	@StreamListener(ReceiveInputChannel.MSG_RECEIVER)
    	public void receiveStr(Message<String> message) { 
    		
    		String s = message.getPayload();
    		logger.info("[receiveStr]消息接收："+s);
    		logger.info("[receiveStr]消息接收处理："+JSONObject.parseObject(s).getString("data"));
    		
    		CdkeyParameter cdkeyParameter = JSON.parseObject(JSONObject.parseObject(s).getString("data"), CdkeyParameter.class);
    		logger.info("CdkeyParameter的参数："+cdkeyParameter.getVersion());
    	}
    
    }

## 2.6 启动类 ##

package com.StreamRabbitMQSelf;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /** * 启动类 * */
    @SpringBootApplication
    public class StreamRabbitMQSelfApplication { 
        public static void main( String[] args ) { 
           SpringApplication.run(StreamRabbitMQSelfApplication.class, args);
        }
    }

## 2.7 其他 ##

### 2.7.1 ParameterUtil ###

/** * */
    package com.StreamRabbitMQSelf.utils.common;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import javax.servlet.http.HttpServletRequest;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    
    
    /** * @author mazhen * 获取post请求中的body内容 */
    public class ParameterUtil { 
    
    	public static String getParametersStr(HttpServletRequest request) throws IOException { 
    		//从request中以"UTF-8"形式获取输入流，避免中文乱码
    		BufferedReader br = new BufferedReader(new InputStreamReader(request.getInputStream(),"UTF-8"));
    		StringBuffer sb = new StringBuffer("");
    		String temp;
    		//一行一行读取并放入到StringBuffer中
    		while((temp = br.readLine()) != null){ 
    			sb.append(temp);
    		}
    		br.close();
    		String acceptjson = sb.toString();
    		return acceptjson;
    	}
    	
    }

### 2.7.2 CdkeyParameter ###

package com.StreamRabbitMQSelf.model;
    
    import java.util.Date;
    
    public class CdkeyParameter { 
        private Long id;
    
        private String telephone;
    
        private String cdkey;
    
        private Short messageStatus;
    
        private Short billStatus;
    
        private Date timeStamp;
    
        private String encode;
    
        private String version;
    
        private Short channel;
    
        private Short flowType;
    
        private Short flowSize;
    
        private Date rechargeTime;
    
        private String memberType;
    
        private String method;
    
        private String serialNumber;
    
        private String orderCodeWo;
    
        private String orderCode;
    
        private Short rechargeState;
    
        private Date createTime;
    
        private Date updateTime;
    
        private String flag1;
    
        private String flag2;
    
        private Integer flag3;
    
        private Integer flag4;
    
        public Long getId() { 
            return id;
        }
    
        public void setId(Long id) { 
            this.id = id;
        }
    
        public String getTelephone() { 
            return telephone;
        }
    
        public void setTelephone(String telephone) { 
            this.telephone = telephone == null ? null : telephone.trim();
        }
    
        public String getCdkey() { 
            return cdkey;
        }
    
        public void setCdkey(String cdkey) { 
            this.cdkey = cdkey == null ? null : cdkey.trim();
        }
    
        public Short getMessageStatus() { 
            return messageStatus;
        }
    
        public void setMessageStatus(Short messageStatus) { 
            this.messageStatus = messageStatus;
        }
    
        public Short getBillStatus() { 
            return billStatus;
        }
    
        public void setBillStatus(Short billStatus) { 
            this.billStatus = billStatus;
        }
    
        public Date getTimeStamp() { 
            return timeStamp;
        }
    
        public void setTimeStamp(Date timeStamp) { 
            this.timeStamp = timeStamp;
        }
    
        public String getEncode() { 
            return encode;
        }
    
        public void setEncode(String encode) { 
            this.encode = encode == null ? null : encode.trim();
        }
    
        public String getVersion() { 
            return version;
        }
    
        public void setVersion(String version) { 
            this.version = version == null ? null : version.trim();
        }
    
        public Short getChannel() { 
            return channel;
        }
    
        public void setChannel(Short channel) { 
            this.channel = channel;
        }
    
        public Short getFlowType() { 
            return flowType;
        }
    
        public void setFlowType(Short flowType) { 
            this.flowType = flowType;
        }
    
        public Short getFlowSize() { 
            return flowSize;
        }
    
        public void setFlowSize(Short flowSize) { 
            this.flowSize = flowSize;
        }
    
        public Date getRechargeTime() { 
            return rechargeTime;
        }
    
        public void setRechargeTime(Date rechargeTime) { 
            this.rechargeTime = rechargeTime;
        }
    
        public String getMemberType() { 
            return memberType;
        }
    
        public void setMemberType(String memberType) { 
            this.memberType = memberType == null ? null : memberType.trim();
        }
    
        public String getMethod() { 
            return method;
        }
    
        public void setMethod(String method) { 
            this.method = method == null ? null : method.trim();
        }
    
        public String getSerialNumber() { 
            return serialNumber;
        }
    
        public void setSerialNumber(String serialNumber) { 
            this.serialNumber = serialNumber == null ? null : serialNumber.trim();
        }
    
        public String getOrderCodeWo() { 
            return orderCodeWo;
        }
    
        public void setOrderCodeWo(String orderCodeWo) { 
            this.orderCodeWo = orderCodeWo == null ? null : orderCodeWo.trim();
        }
    
        public String getOrderCode() { 
            return orderCode;
        }
    
        public void setOrderCode(String orderCode) { 
            this.orderCode = orderCode == null ? null : orderCode.trim();
        }
    
        public Short getRechargeState() { 
            return rechargeState;
        }
    
        public void setRechargeState(Short rechargeState) { 
            this.rechargeState = rechargeState;
        }
    
        public Date getCreateTime() { 
            return createTime;
        }
    
        public void setCreateTime(Date createTime) { 
            this.createTime = createTime;
        }
    
        public Date getUpdateTime() { 
            return updateTime;
        }
    
        public void setUpdateTime(Date updateTime) { 
            this.updateTime = updateTime;
        }
    
        public String getFlag1() { 
            return flag1;
        }
    
        public void setFlag1(String flag1) { 
            this.flag1 = flag1 == null ? null : flag1.trim();
        }
    
        public String getFlag2() { 
            return flag2;
        }
    
        public void setFlag2(String flag2) { 
            this.flag2 = flag2 == null ? null : flag2.trim();
        }
    
        public Integer getFlag3() { 
            return flag3;
        }
    
        public void setFlag3(Integer flag3) { 
            this.flag3 = flag3;
        }
    
        public Integer getFlag4() { 
            return flag4;
        }
    
        public void setFlag4(Integer flag4) { 
            this.flag4 = flag4;
        }
    }

### 2.7.3 TestController ###

/** * */
    package com.StreamRabbitMQSelf.controller;
    
    import java.io.IOException;
    
    import javax.servlet.http.HttpServletRequest;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.StreamRabbitMQSelf.rabbitMQ.service.SendMessage;
    import com.StreamRabbitMQSelf.utils.common.ParameterUtil;
    
    
    
    /** * @author mazhen * */
    @RestController
    public class TestController { 
        
    	/** * 引入日志，注意都是"org.slf4j"包下 */
    	private final static Logger logger = LoggerFactory.getLogger(TestController.class);
    	
    	@Autowired
    	private SendMessage  sendMessage;
    	
    	@RequestMapping(value = "recevieCdkeyFrom",method = RequestMethod.POST)
    	public String recevieCdkeyFrom(HttpServletRequest request){ 
    		
    		String jsonStr = null;
    		try { 
    			jsonStr = ParameterUtil.getParametersStr(request);
    			logger.info("----:"+jsonStr);
    			sendMessage.sendMsgStr(jsonStr);
    		} catch (IOException e) { 
    			logger.error("异常信息："+e);
    			e.printStackTrace();
    			return "IOException:"+e;
    		}
    		return jsonStr;
    	}
    	
    }

# 3 测试 #

*  启动RabbitMQ
 *  启动StreamRabbitMQSelf工程
 *  打开postman

通过postman测试：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]  
控制台打印：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]  
RabbitMQ的交换器：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]  
RabbitMQ的消息队列：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70]: /images/20220418/adafbfd007404756a98b9f38ec9f04bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 1]: /images/20220418/fd7c71c3b4d3478390e98c59a1455069.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 2]: /images/20220418/fa92b598fb114e6c83d2206b20d92866.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 3]: /images/20220418/87dc07b48f7f4303a0915d1b6080aabf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 4]: /images/20220418/04d23c52fc4d4d35b160f08b95fa1d80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70 5]: /images/20220418/5cf5234799374fdd9a6edc33875a7ed6.png
[20191212162452751.png]: /images/20220418/2312f1c25bd24d858fb2a9bf793308bc.png