做美食没有广告的网站,小学网站建设情况汇报,建站之星app,个人购物网站备案目录 ★ 使用topic实现通配符路由代码演示topic通配符类型的Exchange代码演示:ConstantUtilConnectionUtilProducerConsumer01执行结果生产者消费者01消费者02 完整代码#xff1a;ConstantUtilConnectionUtilProducerConsumer01Consumer02pom.xml ★ 使用topic实现通配符路由… 目录 ★ 使用topic实现通配符路由代码演示topic通配符类型的Exchange代码演示:ConstantUtilConnectionUtilProducerConsumer01执行结果生产者消费者01消费者02 完整代码ConstantUtilConnectionUtilProducerConsumer01Consumer02pom.xml ★ 使用topic实现通配符路由
▲ topic类型的Exchange支持在路由key中使用通配符路由key一般由一个或者多个单词组成多个单词之间以“.”分割。
通配符支持星号*和井号# *匹配一个单词。#匹配零个或多个单词。Q1绑定的路由key模式为*.crazyit.*
因此它可以匹配www.crazyit.orgwww.crazyit.cn、edu.crazyit.org等路由key。Q2绑定了两个key模式其中*.org可以匹配crazyit.org、fkjava.org等路由key
但不能匹配www.crazyit.org、www.fkjava.org等*只能匹配一个单词
而edu.#则可匹配edu.crazyit.org、edu.fkjava.org、edu.fkjava、edu.org等#可匹配多个单词。代码演示
需求如图演示通配符
Exchange绑定queue的带通配符的key
ROUTING_PATTERNS {*.crazyit.*, *.org, edu.#};发送消息到Exchange的路由key
ROUTING_KEYS { www.crazyit.org , www.crazyit.cn , edu.crazyit.org , crazyit.org , fkjava.org , edu.crazyit.org , edu.fkjava , edu.org};exchange 和 Q1 绑定的*.crazyit.*符合的路由key有 www.crazyit.org www.crazyit.cn edu.crazyit.org匹配到的key是3 个exchange 和 Q2 绑定的 *.org 和 edu.# 符合的路由key有 *.org -- crazyit.org , fkjava.org , edu.orgedu.# -- edu.crazyit.org , edu.crazyit.org , edu.fkjava , edu.org因为 匹配的key中有两个 edu.org 所以最终匹配到的key 是6个。如图
topic通配符类型的Exchange代码演示:
ConstantUtil
1、先设置一些常量
ConnectionUtil
2、RabbitMQ的相关连接
Producer
3、重点是消息生产者 Consumer01
消费者没啥好说的两个消费者除了指定消费哪个消息队列不同外其他代码都一样。 声明消息队列 生产者和消费者声明消息队列的作用是一样的。 都是为了防止先启动消费者或者先启动生产者时 对应的消息队列还没创建而导致消息被丢弃的问题。 个人理解一般都是生产者需要声明消息队列 消费者声明消息队列的影响不大
执行结果
与期望值一致
生产者 消费者01 exchange 和 Q1 绑定的*.crazyit.*符合的路由key有 www.crazyit.org www.crazyit.cn edu.crazyit.org匹配到的key是3 个消费者02 exchange 和 Q2 绑定的 *.org 和 edu.# 符合的路由key有 *.org -- crazyit.org , fkjava.org , edu.orgedu.# -- edu.crazyit.org , edu.crazyit.org , edu.fkjava , edu.org因为 匹配的key中有两个 edu.org 所以最终匹配到的key 是6个。生产者或消费者声明 Exchagne的作用解释
完整代码
ConstantUtil
package cn.ljh.rabbitmq.util;//常量
public class ConstantUtil
{// ------------topic类型的Exchange需要的相关常量----------public final static String QUEUET01 qt_01;public final static String QUEUET02 qt_02;// topic 通配符类型的 Exchangepublic static final String EXCHANGE_NAME_TOPIC myex03.topic;// Exchange 绑定 Queue 队列的路由key 通配符类型 *匹配一个单词。#匹配零个或多个单词。public static final String[] ROUTING_TOPIC_PATTERNS {*.crazyit.*, *.org, edu.#};// 生产者发送消息给Excahnge携带的路由keypublic static final String[] ROUTING_TOPIC_KEYS { www.crazyit.org, www.crazyit.cn,edu.crazyit.org, crazyit.org, fkjava.org, edu.fkjava.org, edu.fkjava, edu.org};//-------------------------------------------------------// 消息队列的名称public final static String QUEUE01 queue_01;public final static String QUEUE02 queue_02;// Exchange的名称public static final String EXCHANGE_NAME myex02.direct;// 三个路由key定义成一个数组的名称public static final String[] ROUTING_KEYS {info, error, warning};}
ConnectionUtil
package cn.ljh.rabbitmq.util;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;//连接工具
public class ConnectionUtil
{//获取连接的方法public static Connection getConnection() throws IOException, TimeoutException{//创建连接工厂----这个ConnectionFactory源码可以看出有构造器所以直接new一个出来ConnectionFactory connectionFactory new ConnectionFactory();//设置连接信息connectionFactory.setHost(localhost);connectionFactory.setPort(5672);connectionFactory.setUsername(ljh);connectionFactory.setPassword(123456);connectionFactory.setVirtualHost(/); //连接虚拟主机//从连接工厂获取连接Connection connection connectionFactory.newConnection();//返回连接return connection;}
}
Producer
package cn.ljh.rabbitmq.producer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import cn.ljh.rabbitmq.util.ConstantUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;//消息生产者--使用 topic 类型的exchange------就是通配符
public class Producer
{public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接Connection conn ConnectionUtil.getConnection();//2、通过Connection获取Channel。Channel channel conn.createChannel();//3、调用exchangeDeclare()方法声明Exchange--------调用queueDeclare()方法声明队列并完成队列与Exchange的绑定channel.exchangeDeclare(ConstantUtil.EXCHANGE_NAME_TOPIC,/* Exchange名字 */BuiltinExchangeType.TOPIC,/* Exchange 类型--通配符 */true,/* 是否持久化 */false,/* 是否自动栅除 */false,/* 是否为内部的 Exchange */null /* 指定 Exchange 的额外属性 */);//调用queueDeclare()方法声明队列----声明多个消息队列------声明第1个消息队列---------路由key是 通配符 *.crazyit.*channel.queueDeclare(ConstantUtil.QUEUET01, true, false, false, null);//把 Exchange 和 Queue 绑定起来绑定第一个消息队列路由key是通配符channel.queueBind(ConstantUtil.QUEUET01, ConstantUtil.EXCHANGE_NAME_TOPIC,ConstantUtil.ROUTING_TOPIC_PATTERNS[0], /* exchange类型是 topic这里指定路由key -- *.crazyit.*/null /* 指定 Exchange 的额外属性 */);//声明第2个消息队列--------这个exchange绑定这个queue绑定了2个路由key---通配符 *.org, edu.#channel.queueDeclare(ConstantUtil.QUEUET02, true, false, false, null);//用循环为第2个消息队列绑定2个路由keyfor (int i 1; i ConstantUtil.ROUTING_TOPIC_PATTERNS.length; i){//把 Exchange 和 Queue 绑定起来绑定第2个消息队列channel.queueBind(ConstantUtil.QUEUET02,ConstantUtil.EXCHANGE_NAME_TOPIC,ConstantUtil.ROUTING_TOPIC_PATTERNS[i] /* 循环绑定路由key */,null /* 指定 Exchange 的额外属性 */);}//有几个路由key生产者就发送几条消息for (int i 0; i ConstantUtil.ROUTING_TOPIC_KEYS.length; i){//获取路由key用于下面动态指定路由key的代码简洁点String routingKey ConstantUtil.ROUTING_TOPIC_KEYS[i];//要发送的消息String message 生产者发送的第【 (i1) 】条消息的内容;//4、调用Channel 的 basicPublish() 方法发送消息channel.basicPublish(ConstantUtil.EXCHANGE_NAME_TOPIC /* 指定向这个Exchange发送消息 */,routingKey /* 动态指定路由key */,null /*指定额外的消息的属性*/,message.getBytes(StandardCharsets.UTF_8)/*消息体必须是字节数组类型--byte[]*/);System.out.println(生产者发送【 (i1) 】条消息完成发送到Exchange的路由key是routingKey);}//5、关闭资源//关闭通道channel.close();//关闭连接conn.close();}
}
Consumer01
package cn.ljh.rabbitmq.consumer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import cn.ljh.rabbitmq.util.ConstantUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;
// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下
//1创建ConnectionFactory连接工厂设置连接信息再通过ConnectionFactory获取Connection连接。
//2通过Connection获取Channel。
//3根据需要、调用Channel的queueDeclare()方法声明队列 Declare:声明、宣布
// 如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列。
//4调用Channel 的 basicConsume()方法开始处理消息调用该方法时需要传入一个Consumer参数该参数相当于JMS中的消息监听器。//消息消费者1
public class Consumer01
{public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂设置连接信息然后再通过连接工厂获取连接Connection conn ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列// 如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列channel.queueDeclare(ConstantUtil.QUEUET01, /* 声明的队列名 */true, /* 消息队列是否持久化 */false, /* 是否只允许该消息消费者消费该队列的消息独占 */false, /* 是否自动删除 */null /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(ConstantUtil.QUEUET01 /*消费这个消费队列里面的消息*/,true /*消息的确认模式是否自动确认该消息已经被消费完成并返回确认消息给消息队列*/,new DefaultConsumer(channel){//处理消息当这个消息队列收到消息的时候这个方法就会被触发。重写这个方法Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message new String(body, UTF-8);//printf格式化输出函数 %s输出字符串 %n换行System.err.printf(P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息消息内容为%s%n,envelope.getExchange(),envelope.getRoutingKey(),message);}});}
}
Consumer02
package cn.ljh.rabbitmq.consumer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import cn.ljh.rabbitmq.util.ConstantUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者2
public class Consumer02
{public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂设置连接信息然后再通过连接工厂获取连接Connection conn ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列// 如果声明的队列已存在该方法直接获取已有的队列如果声明的队列还不存在该方法将会创建新的队列channel.queueDeclare(ConstantUtil.QUEUET02, /* 声明的队列名 */true, /* 消息队列是否持久化 */false, /* 是否只允许该消息消费者消费该队列的消息独占 */false, /* 是否自动删除 */null /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(ConstantUtil.QUEUET02 /*消费这个名字的消费队列里面的消息*/,true/*消息的确认模式是否自动确认该消息已经被消费完成并返回确认消息给消息队列*/,new DefaultConsumer(channel){//处理消息当这个消息队列收到消息的时候这个方法就会被触发。重写这个方法Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message new String(body, UTF-8);//printf格式化输出函数 %s输出字符串 %n换行System.err.printf(P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息消息内容为%s%n,envelope.getExchange(),envelope.getRoutingKey(),message);}});}}
pom.xml
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcn.ljh/groupIdartifactIdtopic/artifactIdversion1.0.0/versionnametopic/name!-- 属性 --propertiesmaven.compiler.source11/maven.compiler.sourcemaven.compiler.target11/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncodingjava.version11/java.version/properties!-- 依赖 --dependencies!-- RabbitMQ 的依赖库 --dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.13.0/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.16/versionscopecompile/scope/dependency/dependencies/project
