Rabbit MQ使用总结（JAVA）

妖狐艹你老母 2023-05-30 10:46 17阅读 0赞

使用AMPQ client 客户端直接操作，目前框架不是spring的 所以一些常用功能自己封装

**目录**

学习途径：https://blog.csdn.net/vbirdbest/column/info/18247

总结

2. 发布模式

topics：

headers：

fanout：

direct：

3. 消息属性Properties

4. 消息确认机制

5. 队列声明

6. 死信交换机

一般发布者的发布设置比较总要

--------------------

<dependency>
    		    <groupId>com.rabbitmq</groupId>
    		    <artifactId>amqp-client</artifactId>
    		    <version>5.7.3</version>
    </dependency>

# 学习途径：[https://blog.csdn.net/vbirdbest/column/info/18247][https_blog.csdn.net_vbirdbest_column_info_18247] #

[https://www.cnblogs.com/bluesummer/p/8992225.html][https_www.cnblogs.com_bluesummer_p_8992225.html]

# 总结 #

--------------------

## 2. 发布模式 ##

### topics： ###

*  模糊匹配，通过字母配合符号“\*”（一个单词）和“\#”（多个单词）来设置路由键
 *  相同交换机，相同的队列，加入多个路由KEY时 消费端可以通过模糊匹配进路由key获取需要的队列数据，不匹配默认获取全部

### headers： ###

*  以消费端为主要 去匹配 服务器发布的headers。服务器可以多发，但是消费端匹配到的一个都不能少。

### fanout： ###

*  忽略路由键，给所有绑定到交换机上的队列都发送消息,俗称广播

### direct： ###

*  要求路由键完全匹配

## 3. 消息属性Properties ##

*  content\_type ： 消息内容的类型 \[常用 \]
 *  content\_encoding： 消息内容的编码格式
 *  priority： 消息的优先级 根据发布的时消息的重要性
 *  correlation\_id：关联id
 *  reply\_to: 用于指定回复的队列的名称 -> RPC 远程调用时回复是可以带上\[常用 \]
 *  expiration： 消息的失效时间
 *  message\_id： 消息id 拥有确认消息之间是否是同一个，用来消息的回复\[常用 \]
 *  timestamp：消息的时间戳
 *  type： 类型
 *  user\_id: 用户id
 *  app\_id： 应用程序id一般模块名称\[常用 \]
 *  cluster\_id: 集群id

## 4. 消息确认机制 ##

可以自动确认，一收到消息就通知队列删除此消息

可以手动确认，一般和业务逻辑是同一个事务提交 来避免网络导致的无法确认。

## 5. 队列声明 ##

队列声明可以设置持久化，持久化即使服务断开后不会丢失消息

队列声明可以设置无消息是否删除，一般不用

重要参数：

arguments：

Message TTL(x-message-ttl)：设置队列中的所有消息的生存周期(统一为整个队列的所有消息设置生命周期), 也可以在发布消息的时候单独为某个消息指定剩余生存时间,单位毫秒, 类似于redis中的ttl，生存时间到了，消息会被从队里中删除，注意是消息被删除，而不是队列被删除， 特性Features=TTL, 单独为某条消息设置过期时间AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().expiration(“6000”);  
channel.basicPublish(EXCHANGE\_NAME, “”, properties.build(), message.getBytes(“UTF-8”));

Auto Expire(x-expires): 当队列在指定的时间没有被访问(consume, basicGet, queueDeclare…)就会被删除,Features=Exp

Dead letter exchange(x-dead-letter-exchange)： 当队列消息长度大于最大长度、或者过期的等，将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉,Features=DLX

Dead letter routing key(x-dead-letter-routing-key)：将删除的消息推送到指定交换机的指定路由键的队列中去, Feature=DLK

Maximum priority(x-max-priority)：优先级队列，声明队列时先定义最大优先级值(定义最大值一般不要太大)，在发布消息的时候指定该消息的优先级， 优先级更高（数值更大的）的消息先被消费

原文：[https://blog.csdn.net/vbirdbest/article/details/78670550][https_blog.csdn.net_vbirdbest_article_details_78670550]

## 6. 死信交换机 ##

*  一般在发布消息的时候声明死信交换机的路径
 *  在队列声明中配置
 *  Dead letter exchange(x-dead-letter-exchange)： 当队列消息长度大于最大长度、或者过期的等，将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉,Features=DLX
 *  Dead letter routing key(x-dead-letter-routing-key)：将删除的消息推送到指定交换机的指定路由键的队列中去, Feature=DLK
 *  如果消息处理过程中 方法体或者程序校验不通过,可以使用死信队列 人工干预从新处理。
 *  消费端使用Nack 来告诉服务器这个消息 请转发给死信交换机

channel.basicNack(envelope.getDeliveryTag(), false,false);// 程序无法处理 死信队列

# 一般发布者的发布设置比较总要 #

/**
    	 *  根据参数发布消息
    	 * @param appId 当前应用模块名称
    	 * @param message 正常队列名称
    	 * @param exchangeName 队列交换机
    	 * @param exchangeType 交换机类型
    	 * @param routingKey 队列路由
    	 * @param queueName 队列名称
    	 * @param isDeadLetter 是否使用死信队列
    	 * @param dlxExchangeName 死信交换机
    	 * @param dlxExchangeType 死信交换机类型
    	 * @param dlxQueueName 死信队列名称
    	 * @return 返回消息是否发布成功
    	 * @throws Exception
    	 */
    	public static boolean publishMessage(String appId,String message,String exchangeName,String exchangeType,String routingKey,String queueName,Boolean isDeadLetter,String dlxExchangeName,String dlxExchangeType,String dlxQueueName,String dlxRoutingKey) throws Exception{
    	    Connection connection = null;
    	    Channel channel = null;
    	    try {
    	    	RabbitMqProperties mqProperties = new RabbitMqProperties(new Properties());
    			
    		    // 新建一个长连接
    		    connection = mqProperties.getFactory().newConnection();
    		    // 创建一个通道(一个轻量级的连接)
    		    channel = connection.createChannel();
    		    
    		    
    	        Map<String, Object> arguments = null;
    	        if(isDeadLetter){
    	        	/**
    			     * 死信队列配置  ----------------
    			     */
    	        	 // 创建死信交换器和队列
    		        channel.exchangeDeclare(dlxExchangeName, dlxExchangeType, true, false, null);
    		        channel.queueDeclare(dlxQueueName, true, false, false, null);
    		        channel.queueBind(dlxQueueName, dlxExchangeName, dlxRoutingKey);
    		        
    	        	arguments = new HashMap<String, Object>();
    	        	 // 为队列设置队列交换器
    		        arguments.put("x-dead-letter-exchange",dlxExchangeName);
    		        // 设置交换机的路由key
    		        arguments.put("x-dead-letter-routing-key", dlxRoutingKey);
    	        }
    	       
    	
    		    /**
    		     *  正常队列
    		     */
    		    // 交换机的创建
    		    channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
    		    //创建一个持久化、非排他的、非自动删除的队列
    	        channel.queueDeclare(queueName, true, false,false,arguments);
    	        //将交换器与队列通过路由键绑定
    	        channel.queueBind(queueName, exchangeName, routingKey);
    	        
    	        
    		    // 开启发送方确认模式
    		    channel.confirmSelect();
    		    // 发送消息到队列中           
    		   
    		    // 注意：exchange如果不需要写成空字符串，routingKey和队列名称保持一致
    		    AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().appId(appId).contentType(DictUtils.CONTENT_TYPE_TEXT).build();
    		    channel.basicPublish(exchangeName, routingKey, properties, message.getBytes("UTF-8"));
    		    return channel.waitForConfirms();
    
    	    } catch (Exception ex) {
    	        throw new RuntimeException(ex.getMessage());
    	    } finally {
    	        if (channel != null) {
    	            channel.close();
    	        }
    	        if (connection != null) {
    	            connection.close();
    	        }
    	    }
    	}

如果有任何的出错，希望能指正！感谢

[https_blog.csdn.net_vbirdbest_column_info_18247]: https://blog.csdn.net/vbirdbest/column/info/18247
[https_www.cnblogs.com_bluesummer_p_8992225.html]: https://www.cnblogs.com/bluesummer/p/8992225.html
[https_blog.csdn.net_vbirdbest_article_details_78670550]: https://blog.csdn.net/vbirdbest/article/details/78670550