RabbitMQ工作模式5 Topics通配符模式
它也是功能最大的工作模式,它是路由模式的加强版,通过通配符*或者#的方式,更多的匹配路由,交换机类型是topic(通配符),*代表一个单词,#代表零个或者多个单词
模式说明
生产者
package com.wyh.producer; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ生产者 Topic通配符工作模式 * @author: 魏一鹤 * @createDate: 2022-03-23 22:40 **/ import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * producer主要用来发送消息 * * **/ public class Topic_producer { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //5 创建交换机 Exchange exchangeDeclare有很多参数 下面一一说明 //参数1 String exchange, 交换机名称 //参数2 BuiltinExchangeType type, 交换机类型 是一个枚举类型 共有四个值 下面一一介绍 //枚举1 DIRECT("direct"), 定向 //枚举2 FANOUT("fanout"), 扇形(广播) 发送消息到每一个与之绑定队列 //枚举3 TOPIC("topic"), 通配符 //枚举4 HEADERS("headers"); 参数匹配 //参数3 boolean durable, 释放持久化 //参数4 boolean autoDelete, 自动删除 //参数5 boolean internal, 内部使用 一般false //参数6 Map<String, Object> arguments 参数 //交换机名称 String exChangeName="test_exchange_topic"; channel.exchangeDeclare(exChangeName, BuiltinExchangeType.TOPIC,true,false,false,null); //6 创建队列 //队列名称 String queueName1="test_topic_queueName1"; String queueName2="test_topic_queueName2 "; channel.queueDeclare(queueName1,true,false,false,null); channel.queueDeclare(queueName2,true,false,false,null); //7 绑定队列和交换机的关系 让它们两个组合起来 //queueBind有三个参数 //参数1 String queue 队列名称 //参数2 String exchange 交换机名称 //参数3 String routingKey 路由键 绑定规则 //如果交换机的类型为fanout(广播) 那么routingKey设置为空字符串"" //routing key由两部分组成 系统的名称和日志的级别 //需求:所有error级别的日志存数据库,所以order系统的日志存入数据库 channel.queueBind(queueName1,exChangeName,"#.error"); channel.queueBind(queueName1,exChangeName,"order.*"); channel.queueBind(queueName2,exChangeName,"*.*"); //8 发送消息 //定义消息信息 String body="日志信息:张三调用了findAll方法,日志级别为:info"; channel.basicPublish(exChangeName,"order.info",null,body.getBytes()); //9 释放资源 channel.close(); connection.close(); } }
启动生产者查看交换机
消费者1
package com.wyh.consumer; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ消费者Consumer 通配符模式 * @author: 魏一鹤 * @createDate: 2022-03-24 22:08 **/ /** * consumer主要用来消费消息 * * **/ public class Topic_consumer1 { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //队列名称 String queueName1="test_topic_Name1"; String queueName2="test_topic_Name2"; //接收消息 它的参数比较多 下面一一说明 // String queue, 队列名称 // boolean autoAck, 是否自动确认 消费者收到消息会自动告诉MQ它收到了消息 // Consumer callback 回调对象 可以监听一些方法 //consumer本质是一个接口 需要创建它的实现类 Consumer consumer=new DefaultConsumer(channel){ //匿名内部类 重写它的方法 //回调方法 当收到消息后,会自动执行该方法 它有一些参数 //String consumerTag 标识 //Envelope envelope 可以获取一些信息 比如交换机 路由key //AMQP.BasicProperties properties 配置信息 // byte[] body 数据 // @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //System.out.println("consumerTag = " + consumerTag); //System.out.println("exchange = " + envelope.getExchange()); //System.out.println("routingKey = " + envelope.getRoutingKey()); //System.out.println("properties = " + properties); System.out.println("订阅模式消费者1接收的消息 = " + new String(body)); System.out.println("将日志信息存到数据库"); } /** 打印的信息 **/ }; channel.basicConsume(queueName1,true,consumer); //消费者本质是一个监听 所以不要去关闭资源 } }
消费者2
package com.wyh.consumer; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ消费者Consumer 通配符模式 * @author: 魏一鹤 * @createDate: 2022-03-24 22:08 **/ /** * consumer主要用来消费消息 * * **/ public class Topic_consumer2 { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //队列名称 String queueName1="test_topic_Name1"; String queueName2="test_topic_Name2"; //接收消息 它的参数比较多 下面一一说明 // String queue, 队列名称 // boolean autoAck, 是否自动确认 消费者收到消息会自动告诉MQ它收到了消息 // Consumer callback 回调对象 可以监听一些方法 //consumer本质是一个接口 需要创建它的实现类 Consumer consumer=new DefaultConsumer(channel){ //匿名内部类 重写它的方法 //回调方法 当收到消息后,会自动执行该方法 它有一些参数 //String consumerTag 标识 //Envelope envelope 可以获取一些信息 比如交换机 路由key //AMQP.BasicProperties properties 配置信息 // byte[] body 数据 // @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //System.out.println("consumerTag = " + consumerTag); //System.out.println("exchange = " + envelope.getExchange()); //System.out.println("routingKey = " + envelope.getRoutingKey()); //System.out.println("properties = " + properties); System.out.println("订阅模式消费者1接收的消息 = " + new String(body)); System.out.println("将日志信息打印到控制台"); } /** 打印的信息 **/ }; channel.basicConsume(queueName2,true,consumer); //消费者本质是一个监听 所以不要去关闭资源 } }
