摘要:本文将带你了解Exchange服务器之RabbitMQ四种Exchange类型之Topic (Java),希望本文对大家学Exchange有所帮助
Topic类型的Exchange是要进行路由键匹配的。此时需要通过路由键将队列绑定要一个交换器上。规则如下:
符号“#”匹配一个或多个词,例如:“logs.#”能够匹配到“logs.error”、“logs.info.toc”
符号“*”只能匹配一个词。例如:“logs.*” 只能匹配到“logs.error”,不能匹配到“logs.info.toc” 。
如下图:
下面就直接上代码吧!!
Consumer:
package topic;import java.io.IOException;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 TopicConsumer { private static final String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws IOException, TimeoutException { new ExchangeTopic("logs.info"); new ExchangeTopic("logs.*"); new ExchangeTopic("logs.#"); } static class ExchangeTopic{ public ExchangeTopic(final String routingKey) 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, "topic", false, true, null); //队列名称 String queueName = routingKey + ".queue"; //创建队列 channel.queueDeclare(queueName, false, false, true, null); //把队列绑定到路由上 channel.queueBind(queueName, EXCHANGE_NAME, routingKey); System.out.println(" [routingKey = "+ routingKey +"] 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("[routingKey = "+ routingKey +"] Received msg is '" + message + "'"); } }; channel.basicConsume(queueName, true, consumer); } }}
Producer:
package topic;import java.io.IOException;import java.util.concurrent.TimeoutException;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class TopicProducer { private static final String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] argv) throws Exception{ new ExchangeTopic("logs.info", "logs Info test !!"); new ExchangeTopic("logs.error", "logs error test !!"); new ExchangeTopic("logs.error.toc", "logs error toc test !!"); } static class ExchangeTopic{ public ExchangeTopic(String routingKey,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, "topic", false, true, null); channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes()); System.out.println("[routingKey = "+routingKey+"] Sent msg is '" + message + "'"); channel.close(); connection.close(); } }}
结果:
从结果中可以看出,路由键为"logs.info"只匹配到了Producer路由键是"logs.info"的信息,路由键为"logs.*"则匹配到了Producer路由键是"logs.info","logs.error"两条信息,而路由键为"logs.#"则匹配到了Producer路由键是"logs.info","logs.error","logs.error.toc"三条信息。
本文由职坐标整理并发布,希望对同学们有所帮助。了解更多详情请关注职坐标系统运维之Exchange频道!
您输入的评论内容中包含违禁敏感词
我知道了
请输入正确的手机号码
请输入正确的验证码
您今天的短信下发次数太多了,明天再试试吧!
我们会在第一时间安排职业规划师联系您!
您也可以联系我们的职业规划师咨询:
版权所有 职坐标-一站式IT培训就业服务领导者 沪ICP备13042190号-4
上海海同信息科技有限公司 Copyright ©2015 www.zhizuobiao.com,All Rights Reserved.
沪公网安备 31011502005948号