RabbitMQ整合SpringBoot2.x

快来打我* 2022-02-24 13:40 274阅读 0赞

### RabbitMQ整合SpringBoot2.x ###

*  producer配置编写
 *   *  在pom中都引入需要用到的依赖
     *  配置producer的application.properties：
     *  新建好订单实体类Order
     *  在producer包下新建一个订单消息发送类OrderSender
     *  在控制台创建好我们要用的exchange交换机和queue队列并绑定
 *  consumer配置编写
 *   *  pom和application.properties配置
     *  在消费端的entity下也新建好Order实体类
     *  编写消费消息类OrderReceiver

SpringBoot与RabbitMQ集成非常简单，不需要做任何的额外设置，只需要两步即可：

*  step1：引入相关依赖
 *  step2：对application.properties进行配置

首先新建两个项目一个生产者producer一个消费者consumer，项目结构分别如图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]  
这里只做简单的demo演示，所以只用到三个类，实体类Order，消息发送类OrderSender，消息接收类OrderReceiver。

# producer配置编写 #

## 在pom中都引入需要用到的依赖 ##

<dependency>
    	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    
    
    <dependency>
    	<groupId>org.apache.commons</groupId>
    	<artifactId>commons-lang3</artifactId>
    </dependency>
    
    <dependency>
    	<groupId>commons-io</groupId>
    	<artifactId>commons-io</artifactId>
    	<version>2.4</version>
    </dependency>
    
    <dependency>
    	<groupId>com.alibaba</groupId>
    	<artifactId>fastjson</artifactId>
    	<version>1.1.26</version>
    </dependency>
    
    <dependency>
    	<groupId>javax.servlet</groupId>
    	<artifactId>javax.servlet-api</artifactId>
    	<scope>provided</scope>
    </dependency>
    
    <dependency>
    	<groupId>log4j</groupId>
    	<artifactId>log4j</artifactId>
    	<version>1.2.17</version>
    </dependency>

## 配置producer的application.properties： ##

# springboot整合rabbitmq基本配置：
    spring.rabbitmq.addresses=127.0.0.1:5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.connection-timeout=15000
    
    server.servlet.context-path=/
    server.port=8001

## 新建好订单实体类Order ##

package com.herobin.springboot.entity;
    
    import java.io.Serializable;
    
    public class Order implements Serializable{ 
    
    	private static final long serialVersionUID = 8763696299123564381L;
    
    	private String id;
    	
    	private String name;
    	
    	private String messageId; //存储消息发送的唯一标识
    
    	public Order() { 
    	}
    
    	public Order(String id, String name, String messageId) { 
    		this.id = id;
    		this.name = name;
    		this.messageId = messageId;
    	}
    
    	public String getId() { 
    		return id;
    	}
    
    	public void setId(String id) { 
    		this.id = id;
    	}
    
    	public String getName() { 
    		return name;
    	}
    
    	public void setName(String name) { 
    		this.name = name;
    	}
    
    	public String getMessageId() { 
    		return messageId;
    	}
    
    	public void setMessageId(String messageId) { 
    		this.messageId = messageId;
    	}	
    }

## 在producer包下新建一个订单消息发送类OrderSender ##

package com.herobin.springboot.producer;
    
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.herobin.springboot.entity.Order;
    
    @Component
    public class OrderSender { 
    
    	@Autowired
    	private RabbitTemplate rabbitTemplate;
    	
    	public void sendOrder(Order order) throws Exception{ 
    		
    		// 设置消息的唯一id
    		CorrelationData correlationData = new CorrelationData();
    		correlationData.setId(order.getMessageId());
    		// 这里我们传送的exchange为order-exchange，记得要去RabbitMQ的控制台创建好 配置好喝队列的绑定关系
    		rabbitTemplate.convertAndSend("order-exchange", //exchange
    				"order.abcd", //routingKey 
    				order, //object 消息体内容
    				correlationData ); //correlationData 消息唯一id
    	}
    	
    }

## 在控制台创建好我们要用的exchange交换机和queue队列并绑定 ##

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

点击我们的交换机与我们队列绑定  
![在这里插入图片描述][20190404111640676.png]  
这里`.#` 和`.*`是有区别的，前者`.#`适用与各种配置场景（多点），如order.a.b.c，而后者`.*`只能配置一个点的，如order.a，order.abc，无法识别order.abc.def这种两个点的场景。

现在我们可以去测试方法里进行测试了

启动测试，测试成功。

去控制台queue队列下我们可以看到已经有了一条队列  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 3]

说明我们的消息队列发送（生产消息）成功。

# consumer配置编写 #

## pom和application.properties配置 ##

consumer的pom用到的和producer都是一样的，application.properties只要将端口号变更一下，再加入消费端的配置就好了

# springboot整合rabbitmq基本配置 
    spring.rabbitmq.addresses=127.0.0.1:5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.connection-timeout=15000
    
    # springboot整合rabbitmq消费端配置
    spring.rabbitmq.listener.simple.concurrency=5
    spring.rabbitmq.listener.simple.max-concurrency=10
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    spring.rabbitmq.listener.simple.prefetch=1
    
    server.servlet.context-path=/
    server.port=8002

## 在消费端的entity下也新建好Order实体类 ##

## 编写消费消息类OrderReceiver ##

package com.herobin.springboot.consumer;
    
    import java.util.Map;
    
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    import com.herobin.springboot.entity.Order;
    import com.rabbitmq.client.Channel;
    
    @Component
    public class OrderReceiver { 
    
    	@RabbitListener(bindings = @QueueBinding(
    				value = @Queue(value = "order-queue", durable = "true"),
    				exchange = @Exchange(name = "order-exchange", durable = "true", type = "topic"),
    				key = "order.*"
    			)
    	)
    	@RabbitHandler
    	public void onOrderMessage(@Payload Order order,
    			@Headers Map<String, Object> headers,
    			Channel channel) throws Exception{ 
    		// 消费者操作
    		System.err.println("----------收到消息，开始消费----------");
    		System.err.println("订单ID：" + order.getId());
    		Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
    		// ACK 确认签收消息
    		channel.basicAck(deliveryTag, false);
    	}
    	
    }

这里的`@RabbitListener`是非常强大的即使我们没有在控制台创建order-exchange和order-queue在执行了启动类之后也会帮我们自动创建好。

启动我们的consumer程序，清空控制台（方便直观显示消息信息），再去启动producer的测试方法发送消息

回到consumer的console发现输出了消息  
![在这里插入图片描述][20190404112228645.png]  
至此，实现了rabbitmq整合springboot简单的消息发送与接收。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70]: /images/20220224/4e193fd4331a4ae9a28ce0c4d8ee2d2e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 1]: /images/20220224/a75026cbeb634395a72106118365517e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 2]: /images/20220224/354198125bb441b29304fa46c1271183.png
[20190404111640676.png]: /images/20220224/792e17693f6d414fa27180a2765eb8de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70 3]: /images/20220224/8ffd71dd517c45dcb6ab0fc506d12d6c.png
[20190404112228645.png]: /images/20220224/af3b6e13c2844dafa106480c7c344e4b.png