云诺说 - 小程序定制 - 广告营销
当前位置: RabbitMQ > RabbitMQ四种Exchange类型之Headers(Java)

RabbitMQ四种Exchange类型之Headers(Java)

2019-07-15 05:47 分类:RabbitMQ 作者:云诺 阅读(147)

版权声明:本文为博主原创文章,如果转载请给出原文链接:http://doofuu.com/article/4156157.html

Headers 类型的Exchanges是不处理路由键的,而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对;当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配;如果完全匹配则消息会路由到该队列,否则不会路由到该队列。headers属性是一个键值对,可以是Hashtable,键值对的值可以是任何类型。而fanout,direct,topic 的路由键都需要要字符串形式的。

匹配规则x-match有下列两种类型:

  • x-match = all   :表示所有的键值对都匹配才能接受到消息

  • x-match = any :表示只要有键值对匹配就能接受到消息

不过headers比较少用到,下面是headers的官方说明文档:

A headers exchange is designed to for routing on multiple attributes that are more easily expressed as message headers than a routing key. Headers exchanges ignore the routing key attribute. Instead, the attributes used for routing are taken from the headers attribute. A message is considered matching if the value of the header equals the value specified upon binding.

It is possible to bind a queue to a headers exchange using more than one header for matching. In this case, the broker needs one more piece of information from the application developer, namely, should it consider messages with any of the headers matching, or all of them? This is what the "x-match" binding argument is for. When the "x-match" argument is set to "any", just one matching header value is sufficient. Alternatively, setting "x-match" to "all" mandates that all the values must match.

Headers exchanges can be looked upon as "direct exchanges on steroids". Because they route based on header values, they can be used as direct exchanges where the routing key does not have to be a string; it could be an integer or a hash (dictionary) for example.

下面贴出headers的事例代码!!

Consumer:

package headers;
 
import java.io.IOException;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.TimeoutException;
 
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
 
 
public class HeadersConsumer {
 
	private static final String EXCHANGE_NAME 	= "exchange_headers";
	private static final String QUEUE_NAME 		= "headers_test_queue";
	public static void main(String[] argv) throws IOException, TimeoutException  {
		new ExchangeHeaders();
	}
 
	static class ExchangeHeaders{
		public  ExchangeHeaders() throws IOException, TimeoutException {
			ConnectionFactory factory = new ConnectionFactory();
			//rabbitmq监听IP
			factory.setHost("192.168.249.128");
			//rabbitmq监听默认端口
			factory.setPort(5672);
			//设置访问的用户
			factory.setUsername("test");
			factory.setPassword("test");
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			//声明路由名字和类型
			channel.exchangeDeclare(EXCHANGE_NAME, "headers", false, true, null);
			//创建队列
			channel.queueDeclare(QUEUE_NAME, false, false, true, null);
			
			//设置消息头键值对信息
			Map<String, Object> headers = new Hashtable<String,Object>();
			//这里x-match有两种类型
			//all:表示所有的键值对都匹配才能接受到消息
			//any:表示只要有键值对匹配就能接受到消息
			headers.put("x-match", "any");
			headers.put("name", "jack");
			headers.put("age" , 31);
			
			//把队列绑定到路由上并指定headers
			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", headers);
			System.out.println(" Waiting for msg....");
			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope,
						AMQP.BasicProperties properties, byte[] body) throws IOException {
					String message = new String(body, "UTF-8");
					
					System.out.println("Received msg is '" + message + "'");
				}
			};
			channel.basicConsume(QUEUE_NAME, true, consumer);
		}
 
	}
 
}

Producer:

package headers;
import java.io.IOException;
import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.TimeoutException;
 
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
 
public class HeadersProducer {
 
	private static final String EXCHANGE_NAME = "exchange_headers";
 
	public static void main(String[] argv) throws Exception{
		new ExchangeHeaders("exchanges type headers test msg~");
	}
	
	static class ExchangeHeaders{
		public ExchangeHeaders(String message) throws IOException, TimeoutException{
			ConnectionFactory factory = new ConnectionFactory();
			//rabbitmq监听IP
			factory.setHost("192.168.249.128");
			//rabbitmq监听默认端口
			factory.setPort(5672);
			//设置访问的用户
			factory.setUsername("test");
			factory.setPassword("test");
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
 
			//声明路由名字和类型
			channel.exchangeDeclare(EXCHANGE_NAME, "headers", false, true, null);
			
			//设置消息头键值对信息
			Map<String, Object> headers = new Hashtable<String, Object>();
			headers.put("name", "jack");
			headers.put("age", 30);
			Builder builder = new Builder();
			builder.headers(headers);
			
			channel.basicPublish(EXCHANGE_NAME, "", builder.build(), message.getBytes());
			System.out.println("Sent msg is '" + message + "'");
 
			channel.close();
			connection.close();
			
		}
	}
 
}

运行结果:

注意如果将Consumer中的headers.put("x-match", "any");改成headers.put("x-match", "all"); 则不会接受到消息!!

 

祝生活愉快!

「创作不易,你的支持是本站持续更新最大的动力!」

赞(0) 打赏

谢谢你请我喝奶茶*^_^*

支付宝
微信
1

谢谢你请我喝奶茶*^_^*

支付宝
微信

上一篇:

下一篇:

共有 0 条评论 - RabbitMQ四种Exchange类型之Headers(Java)

博客简介

云诺说是一个致力于分享互联网编程技术交流、小程序开发、小程序源码分享、网络推广引流服务、推荐优质的引流渠道、广告开户、竞价推广等行业的IT技术网站,希望你能在本站得到你所需要的信息。

Q Q :2730094141

微信 :LGY17666

职业 :小程序开发、网络推广

现居 :广东省-广州市-天河区

友情链接

欢迎与本博客交换友情链接,本博客对交换链接的网站没有要求。如果您是本博客的友情链接网站,在遇到网站运行问题时,可以随时联系,我们将免费提供技术类支持! 申请交换友链

站点统计

  • 文章总数:76 篇
  • 草稿数目:0 篇
  • 分类数目:14 个
  • 独立页面:86 个
  • 评论总数:0 条
  • 访问总量: 12055次
  • 最近更新:2019年11月21日