rabbitmq实战系列2--springboot整合rabbitmq的插件实现延迟队列(2)

利用插件rabbitmq_delayed_message_exchange插件方式

1下载地址
2下载.ez文件,解压后放在rabbitmq的Plugins文件里
3命令行输入rabbitmq-plugins enable rabbitmq_delayed_message_exchange
4重启rabbitmq

编写配置

import java.util.HashMap;
import java.util.Map;

@Configuration
public class ttlq {
public static final String QUEUE_NAME = "delayed.queue";
public static final String EXCHANGE_NAME = "delayedExchange";

@Bean
public Queue queue() {
return new Queue(ttlq.QUEUE_NAME);
}

//配置默认的交换机
@Bean
CustomExchange customExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
//参数二为类型：必须是x-delayed-message
return new CustomExchange(ttlq.EXCHANGE_NAME, "x-delayed-message", true, false, args);
}

//绑定队列到交换器
@Bean
Binding binding(Queue queue, CustomExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ttlq.QUEUE_NAME).noargs();
}
}

生产者

@Component
public class ttltest {
@Autowired
RabbitTemplate template;
public void send1(List<Integer> list){
SimpleDateFormat format=new SimpleDateFormat("hh:mm:ss");
System.out.println("发送时间:"+ format.format(new Date()));
template.convertAndSend("delayedExchange","delayed.queue",list,message ->{
message.getMessageProperties().setHeader("x-delay", 20000);//注意和前面得不一样
return message;
} );
}
public void send2(String str){
SimpleDateFormat format=new SimpleDateFormat("hh:mm:ss");
System.out.println("第二个消息得发送时间:"+ format.format(new Date()));
template.convertAndSend("delayedExchange","delayed.queue",str,message -> {
message.getMessageProperties().setHeader("x-delay",2000);
return message;
});
}
}

消费者

@Component
@RabbitListener(queues = "delayed.queue")
public class ttlConsumer2 {
@RabbitHandler
public void hand(List<Integer>list){
SimpleDateFormat format=new SimpleDateFormat("hh:mm:ss");
System.out.println("接收到了第一个消息,现在时间是:"+ format.format(new Date()));
System.out.println(list.toString());
}
@RabbitHandler
public void hand1(String str){
SimpleDateFormat format=new SimpleDateFormat("hh:mm:ss");
System.out.println("接收到了第二个消息,现在时间是:"+format.format(new Date()));
System.out.println(str);
}
}

测试类:

@Test
public void test1(){
List<Integer>list = new LinkedList<>();
list.add(1);
list.add(2);
list.add(3);
list.add(4);
ttltest.send1(list);
}
@Test
public void test2(){
ttltest.send2("hello world");
}

结果:
可以看到尽管第一个消息过期时间更长,但是第二个消息先被处理,没有发生队列阻塞现象

• 点赞
• 收藏
• 分享
• 文章举报