1、依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2、rabbitmq鏈接配置
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: wq
password: qifeng
virtual-host: /
#開啟ack
listener:
direct:
acknowledge-mode: manual
prefetch: 1 # 限制一次拉取訊息的數量
simple:
acknowledge-mode: manual #采取手動應答
#concurrency: 1 # 指定最小的消費者數量
#max-concurrency: 1 #指定最大的消費者數量
retry:
enabled: true # 是否支持重試
3、生產者
- 宣告交換機、佇列、系結關于
package com.wanqi.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Configuration
public class RabbitMQConfig {
public static final String EXCHANGE_NAME = "boot_topic_exchange";
public static final String QUEUE_NAME = "boot_queue";
//交換機
@Bean("bootExchange")
public Exchange bootExchange(){
return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true)
.build();
}
//
@Bean("bootQueue")
public Queue bootQueue(){
return QueueBuilder.durable(QUEUE_NAME).build();
}
//佇列和交換機系結關系
/*
知道哪個佇列
知道哪個交換機
知道routing key
*/
@Bean
public Binding bindingQueueExchange(@Qualifier("bootQueue") Queue queue,
@Qualifier("bootExchange") Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
}
- 發送訊息
package com.wanqi;
import com.wanqi.mq.RabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class SpringbootMqProducersApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads() {
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "boot.test", "boot mq hello~~~");
}
}
4、消費者
package com.wanqi.listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Component
public class RabbitMQListener {
@RabbitListener(queues = {"boot_queue"})
public void listenerQueue(Object msg, Message message, Channel channel) {
final long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
System.out.println(msg.toString());
System.out.println(new String(message.getBody()));
// int x = 3/0;
channel.basicAck(deliveryTag, true);
} catch (Exception e) {
try {
channel.basicNack(deliveryTag, true, true);
} catch (IOException ex) {
throw new RuntimeException(ex);
}
}
}
}
5、死信佇列
訊息成為死信的三種情況
1、佇列訊息長度到達限制；
2、消費者拒接消費訊息，并且不重回佇列；
3、原佇列存在訊息過期設定，訊息到達超時時間未被消費；
5.1、宣告交換機、佇列、系結佇列
- topic模式
package com.wanqi.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Configuration
public class QDRabbitMQConfig {
//交換機名稱
public static final String ITEM_EXCHANGE = "item_exchange";
public static final String DEAD_EXCHANGE = "dead_exchange";
//佇列名稱
public static final String ITEM_QUEUE = "item_queue";
public static final String DEAD_QUEUE = "dead_queue";
//宣告業務交換機
@Bean("itemExchange")
public Exchange itemExchange(){
return ExchangeBuilder.topicExchange(ITEM_EXCHANGE).durable(true).build();
}
//宣告死信交換機
@Bean("deadExchange")
public Exchange deadExchange(){
return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).durable(true).build();
}
/**
* 宣告普通佇列，設定佇列訊息過期時間，佇列長度，系結的死信佇列
*/
@Bean("itemQueue")
public Queue itemQueue(){
return QueueBuilder
.durable(ITEM_QUEUE)
// 佇列訊息過期時間
.ttl(10000)
// 佇列長度
.maxLength(15)
// 宣告當前佇列系結的死信交換機
.deadLetterExchange(DEAD_EXCHANGE)
// 宣告當前佇列死信轉發的路由key
.deadLetterRoutingKey("infoDead.haha")
.build();
}
/**
* 死信佇列，消費者需要監聽的佇列
*/
@Bean("deadQueue")
public Queue deadQueue(){
return QueueBuilder
.durable(DEAD_QUEUE)
.build();
}
//系結佇列和交換機（業務）
@Bean
public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
@Qualifier("itemExchange") Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("infoRouting.#").noargs();
}
//系結佇列和交換機（死信）
@Bean
public Binding deadQueueExchange(@Qualifier("deadQueue") Queue queue,
@Qualifier("deadExchange") Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("infoDead.#").noargs();
}
}
5.2、發送測驗訊息
package com.wanqi;
import com.wanqi.mq.RabbitMQConfig;
import com.wanqi.mq.RabbitMQConfig2;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class SpringbootMqProducersApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
// 原佇列存在訊息過期設定，訊息到達超時時間未被消費；
@Test
void contextLoads3() {
MessageProperties messageProperties = new MessageProperties();
// 設定過期時間，單位：毫秒
messageProperties.setExpiration("5000");
byte[] msgBytes = "rabbitmq ttl message ...".getBytes();
Message message = new Message(msgBytes, messageProperties);
//發送訊息
rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",message);
System.out.println("發送訊息成功");
}
// 模擬佇列訊息長度到達限制
@Test
void contextLoads4() {
for (int i = 0; i < 20; i++) {
rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",i + "---message");
}
}
}
5.3、消費者
package com.wanqi.listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Component
public class RabbitMQListener {
@RabbitListener(queues = {"dead_queue"})
public void listenerQueue2(Object msg, Message message, Channel channel) {
final long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
System.out.println(msg.toString());
channel.basicAck(deliveryTag, true);
} catch (Exception e) {
try {
channel.basicNack(deliveryTag, true, true);
} catch (IOException ex) {
throw new RuntimeException(ex);
}
}
}
}
6、延遲佇列（TTL + 死信佇列）
6.1、宣告交換機、佇列、系結佇列
- direct模式
package com.wanqi.mq;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Configuration
public class TtlQueueConfig {
//普通交換機名稱
public static final String X_CHANGE = "X_Exchange";
//死信交換機名稱
public static final String Y_DEAD_CHANGE = "Y_Exchange";
//普通佇列
public static final String QUEUE_A = "QA_QUEUE";
public static final String QUEUE_B = "QB_QUEUE";
//死信佇列
public static final String DEAD_QUEUE_D = "QD_QUEUE";
//宣告普通交換機
@Bean("xExchange")
public DirectExchange xExchange() {
return ExchangeBuilder.directExchange(X_CHANGE).durable(true)
.build();
}
//宣告死信交換機
@Bean("yExchange")
public DirectExchange yExchange() {
return ExchangeBuilder.directExchange(Y_DEAD_CHANGE).durable(true)
.build();
}
/**
* 宣告佇列,延遲10秒
*/
@Bean("queueA")
public Queue queueA() {
return QueueBuilder.durable(QUEUE_A)
.deadLetterExchange(Y_DEAD_CHANGE) //死信交換機
.deadLetterRoutingKey("YD") //死信RoutingKey
.ttl(10000) //訊息過期時間
.build();
}
/**
* 宣告佇列，延遲40秒
*
* @return
*/
@Bean("queueB")
public Queue queueB() {
return QueueBuilder.durable(QUEUE_B)
.deadLetterExchange(Y_DEAD_CHANGE) //死信交換機
.deadLetterRoutingKey("YD") //死信RoutingKey
.ttl(40000) //訊息過期時間
.build();
}
/**
* 死信佇列，消費者需要監聽的佇列
*/
@Bean("queueD")
public Queue queueD() {
return QueueBuilder.durable(DEAD_QUEUE_D).build();
}
//系結 X_CHANGE系結queueA
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange) {
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
//系結 X_CHANGE系結queueB
@Bean
public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
//系結 Y_CHANGE系結queueD
@Bean
public Binding queueDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange) {
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
6.2、發送訊息
@Test
void contextLoads5() {
String message = "延遲佇列訊息";
rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XA", "TTL=10s的佇列：" + message);
rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XB", "TTL=40s的佇列：" + message);
}
6.3、消費
package com.wanqi.listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Date;
/**
* @Description TODO
* @Version 1.0.0
* @Date 2022/11/19
* @Author wandaren
*/
@Component
public class RabbitMQListener {
@RabbitListener(queues = "QD_QUEUE")
public void receiveMessage(Object msg,Message message, Channel channel){
final long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
System.out.println(msg.toString());
System.out.print("當前時間: " + new Date());
System.out.println(" 收到死信佇列的訊息：" + msg);
channel.basicAck(deliveryTag, true);
} catch (Exception e) {
try {
channel.basicNack(deliveryTag, true, true);
} catch (IOException ex) {
throw new RuntimeException(ex);
}
}
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/536208.html
標籤：其他
上一篇：class位元組碼檔案結構