一、基于docker搭建rabbitMQ(带有管理界面)
1.docker pull rabbitmq:3-management
2.docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
注:5672为通信端口,
15672为后台管理端口
二、springbooot基层rabbitMQ
1.基于字符串传输
生产者:
1)引包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
2)配置mq
# 配置rabbitMQspring:
spring.rabbitmq.addresses=自己搭建的mq的IP和端口
spring.rabbitmq.username=账户
spring.rabbitmq.password=密码
spring.rabbitmq.virtual-host= /
spring.rabbitmq.connection-timeout= 15000
3)发送bean编写
@Component
public class OrderSender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendOrder(String message){
// 交换机
String exchange = "order-exchange";
// 路由规则
String routingKey = "order.test";
CorrelationData correlationData = new CorrelationData();
correlationData.setId(UUID.randomUUID().toString());
// 第一个参数:生产者要发送的交换机
// 第二个参数:消息路由规则
// 第三个参数:消息体
// 第四个参数:消息唯一id
rabbitTemplate.convertAndSend(exchange,routingKey, message, correlationData);
}
}
4)测试消息发送
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ProductorApplication.class)
public class OrderSenderTest {
@Autowired
private OrderSender orderSender;
@Test
public void sendOrderTest(){
orderSender.sendOrder(order);
}
}
消费者
1)引包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
2)配置mq
# 配置rabbitMQspring:
spring.rabbitmq.addresses=自己搭建的mq的IP和端口
spring.rabbitmq.username=账户
spring.rabbitmq.password=密码
spring.rabbitmq.virtual-host= /
spring.rabbitmq.connection-timeout= 15000
3)监听编写
@Component
public class OrderConsumer {
/**
* @RabbitListener 可以帮我们创建 exchange、binding、queue
* exchange : 交换机信息配置
* key : 路由 key
* value : 监听的队列信息配置
* @param order
* @param headers
* @param channel
* @throws IOException
*/
@RabbitHandler
@RabbitListener(
bindings = @QueueBinding(
exchange = @Exchange(name = "order-exchange", durable = "true", type = "topic"),
key = "order.#",
value = @Queue(value = "order-queue", durable = "true")
)
)
public void onOrderMessage(@Payload Order order, @Headers Map<String, Object> headers, Channel channel) throws IOException {
// 消费者操作
System.err.println("--收到消息,开始消费--");
// System.err.println("the data what " + JSON.parseObject(order, Order.class).toString());
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手工签收
channel.basicAck(deliveryTag, false);
}
基于对象传输
与字符串类似,
convertAndSend(exchange,routingKey, message, correlationData);方法message可以接受object,但是要注意,如果是对象的话要保证发送和接收类路径一样,这个是很容易猜到的坑!
原文:https://www.cnblogs.com/mayWorldPeace/p/14266129.html