RabbitMQ之业务场景：动态创建，删除队列工具类（一）

RabbitConfig.java 配置类

import java.util.HashMap;
import java.util.Map;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 配置rabbitMQ* @author Rayson517**/
@Configuration
public class RabbitConfig {@Beanpublic RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){return new RabbitAdmin(connectionFactory);}@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);//数据转换为json存入消息队列rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());return rabbitTemplate;}
}

RabbitUtil.java工具类

import java.util.Date;
import java.util.UUID;import javax.annotation.Resource;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AbstractExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.SerializationUtils;@Configuration
public class RabbitUtil {private static final Logger logger = LoggerFactory.getLogger(RabbitUtil.class);private final RabbitAdmin rabbitAdmin;private final RabbitTemplate rabbitTemplate;@Autowiredpublic RabbitUtil(RabbitAdmin rabbitAdmin, RabbitTemplate rabbitTemplate){this.rabbitAdmin = rabbitAdmin;this.rabbitTemplate = rabbitTemplate;}/*** 转换Message对象* @param messageType 返回消息类型 MessageProperties类中常量* @param msg* @return*/public Message getMessage(String messageType, Object msg){MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType(messageType);Message message = new Message(msg.toString().getBytes(),messageProperties);return message;}/*** 有绑定Key的Exchange发送* @param routingKey* @param msg*/public void sendMessageToExchange(TopicExchange topicExchange, String routingKey, Object msg){Message message = getMessage(MessageProperties.CONTENT_TYPE_JSON,msg);rabbitTemplate.send(topicExchange.getName(), routingKey, message);}/*** 没有绑定KEY的Exchange发送* @param exchange* @param msg*/public void sendMessageToExchange(TopicExchange topicExchange, AbstractExchange exchange, String msg){addExchange(exchange);logger.info("RabbitMQ send "+exchange.getName()+"->"+msg);rabbitTemplate.convertAndSend(topicExchange.getName(),msg);}/*** 给queue发送消息* @param queueName* @param msg*/public void sendToQueue(String queueName, String msg){sendToQueue(DirectExchange.DEFAULT, queueName, msg);}/*** 给direct交换机指定queue发送消息* @param directExchange* @param queueName* @param msg*/public void sendToQueue(DirectExchange directExchange, String queueName, String msg){Queue queue = new Queue(queueName);addQueue(queue);Binding binding = BindingBuilder.bind(queue).to(directExchange).withQueueName();rabbitAdmin.declareBinding(binding);//设置消息内容的类型，默认是 application/octet-stream 会是 ASCII 码值rabbitTemplate.convertAndSend(directExchange.getName(), queueName, msg);}/*** 给queue发送消息* @param queueName* @param msg*/public String receiveFromQueue(String queueName){return receiveFromQueue(DirectExchange.DEFAULT, queueName);}/*** 给direct交换机指定queue发送消息* @param directExchange* @param queueName* @param msg*/public String receiveFromQueue(DirectExchange directExchange, String queueName){Queue queue = new Queue(queueName);addQueue(queue);Binding binding = BindingBuilder.bind(queue).to(directExchange).withQueueName();rabbitAdmin.declareBinding(binding);String messages = (String)rabbitTemplate.receiveAndConvert(queueName);System.out.println("Receive:"+messages);return messages;}/*** 创建Exchange* @param exchange*/public void addExchange(AbstractExchange exchange){rabbitAdmin.declareExchange(exchange);}/*** 删除一个Exchange* @param exchangeName*/public boolean deleteExchange(String exchangeName){return rabbitAdmin.deleteExchange(exchangeName);}/*** Declare a queue whose name is automatically named. It is created with exclusive = true, autoDelete=true, and* durable = false.* @return Queue*/public Queue addQueue(){return rabbitAdmin.declareQueue();}/*** 创建一个指定的Queue* @param queue* @return queueName*/public String addQueue(Queue queue){return rabbitAdmin.declareQueue(queue);}/*** Delete a queue.* @param queueName the name of the queue.* @param unused true if the queue should be deleted only if not in use.* @param empty true if the queue should be deleted only if empty.*/public void deleteQueue(String queueName, boolean unused, boolean empty){rabbitAdmin.deleteQueue(queueName,unused,empty);}/*** 删除一个queue* @return true if the queue existed and was deleted.* @param queueName*/public boolean deleteQueue(String queueName){return rabbitAdmin.deleteQueue(queueName);}/*** 绑定一个队列到一个匹配型交换器使用一个routingKey* @param queue* @param exchange* @param routingKey*/public void addBinding(Queue queue ,TopicExchange exchange,String routingKey){Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);rabbitAdmin.declareBinding(binding);}/*** 绑定一个Exchange到一个匹配型Exchange 使用一个routingKey* @param exchange* @param topicExchange* @param routingKey*/public void addBinding(Exchange exchange,TopicExchange topicExchange,String routingKey){Binding binding = BindingBuilder.bind(exchange).to(topicExchange).with(routingKey);rabbitAdmin.declareBinding(binding);}/*** 去掉一个binding* @param binding*/public void removeBinding(Binding binding){rabbitAdmin.removeBinding(binding);}
}

然后在需要的地方注入这个RabbitUtil工具类就可以使用了

RabbitMQ之业务场景：动态创建，删除队列工具类（一）相关推荐

安卓创建第一个工具类
安卓创建工具类 1.运用IOC框架简化代码注解 (Annotation):@Class 给代码看的处理代码逻辑 2. 创建自己的注解 @Target() ElementType.FIELD 放在属性 ...
Mybatis的动态创建删除表
Mybatis中可以使用JSTL标签动态删除表 Mapper void deleteTable(@Param("tableName") String tableName); Ma ...
JUC（3）List、Set、Map集合线程安全Callable创建线程三大工具类：CountDownLatch减法计数器、CyclicBarrier加法计数器、Semaphore计数信号量
1. List集合线程安全 CopyOnWriteArrayList是线程安全的集合: ArrayList是线程不安全的集合: Vector是线程安全的集合(不推荐使用) 1.1 解决ArrarLis ...
MySQL 新增表中的数据为另外一个或多个表的数据（业务场景：创建关系表，复制旧表数据到新表）
问题昨天同学问了我个问题我想写个sql脚本,循环插入数据,数据是另一个表中的a.b字段 A表中的a.b字段数据,循环加入B表中a.b字段.就是这个意思我问他为什么不用Java写,他说是数据维护, ...
消息队列工具类(MSMQ)
所要做的是简化msmq的调用代码以及做到可替代性,实现后,调用消息队列代码变为如下所示: QueueService srv = QueueService.Instance();//检查存储DTO1的队 ...
Mybatis动态创建表
背景: 项目中涉及存储数据量过大,会导致数据读取慢等问题,所以考虑动态创建表来存储数据. ↓ 这里场景动态创建表,而且字段是不固定的.表名和字段对应关系存储在单张表中! 解释: 创建表涉及到数据类型. ...
Drools动态创建规则文件并动态调用
目录 1.编写规则内容,如下图 2.创建KieSession 3.根据规则名称货主匹配规则进行执行 3.1.根据规则名称完全匹配方式执行 3.2.根据规则名称前缀匹配方式执行 1.编写规则内容,如下图 ...
Android动态加载黑科技动态创建Activity模式
基本信息 Author:kaedea GitHub:android-dynamical-loading 代理Activity模式的限制还记得我们在代理Activity模式里谈到启动插件APK里的Ac ...
Python基础笔记_Day12_Python元类、type动态创建类、Python动态创建方法、Python运算符、Python发邮件、短信
Day12_Python元类.type动态创建类.Python动态创建方法.Python运算符重载.Python发邮件.短信 12.01_Python语言基础(类对象)(熟悉) 12.02_Pytho ...
面对不同的业务场景，选择零码还是低码？
摘要: 如果你尚未完全了解低代码和零代码这两条不尽相同的道路,相信这篇文章能帮助你建立初步的认识. 本文分享自华为云社区<面对不同的业务场景,选择零码还是低码?>,作者:由月 . 作者简介 ...

RabbitMQ之业务场景：动态创建，删除队列工具类（一）

RabbitMQ之业务场景：动态创建，删除队列工具类（一）相关推荐

最新文章

热门文章