标签：实现 cpu width mic att 服务器 prope ring bin
/* 发布/订阅使用的交换机类型为: fanout */ public class Producer { //交换机名称 static final String FANOUT_EXCHAGE = "fanout_exchange"; //队列名称 static final String FANOUT_QUEUE_1 = "fanout_queue_1"; //队列名称 static final String FANOUT_QUEUE_2 = "fanout_queue_2"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); /** * 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型，fanout、topic、direct、headers(不常用) */ channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null); channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, ""); channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, ""); //发送消息 for (int i = 0; i < 10; i++) { String message = "RabbitMQ" + Math.random(); /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes()); } System.out.println("已发送消息"); //关闭资源 channel.close(); connection.close(); } }
public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); //声明队列1 channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, ""); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("消费者1-接收到的消息为：" + new String(body)); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer); } } public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); //声明队列2 channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, ""); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("消费者2-接收到的消息为：" + new String(body)); } }; //监听消息 channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer); } }
/** * 路由模式的交换机类型为：direct */ public class Producer { //交换机名称 static final String DIRECT_EXCHAGE = "direct_exchange"; //队列名称 static final String DIRECT_QUEUE_INSERT = "direct_queue_insert"; //队列名称 static final String DIRECT_QUEUE_UPDATE = "direct_queue_update"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); //声明队列 channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null); channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null); //队列绑定交换机 /** * 参数1: 队列名 * 参数2: 交换机名 * 参数3: Routing Key */ channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert"); channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update"); // 发送信息 String message = "routing key为insert"; /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes()); System.out.println("已发送消息: " + message); // 发送信息 message = "routing key为update"; channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes()); System.out.println("已发送消息: " + message); // 关闭资源 channel.close(); connection.close(); } }
public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); //创建队列 channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert"); //创建消费者, 并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "utf-8")); } }; channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer); } } public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null); channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update"); DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "utf-8")); } }; channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer); } }
item.* 只能匹配item.insert item.# 能匹配item.insert, item.insert.abc
public class Producer { //交换机名称 static final String TOPIC_EXCHAGE = "topic_exchange"; //队列名称 static final String TOPIC_QUEUE_1 = "topic_queue_1"; //队列名称 static final String TOPIC_QUEUE_2 = "topic_queue_2"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); //发送信息 String message = "Topic: routing key为item.insert"; channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes()); message = "Topic: routing key为item.update"; channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes()); message = "Topic: routing key为item.delete"; channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes()); channel.close(); connection.close(); } }
public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update"); channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body, "utf-8")); } }; //监听消息 channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer); } } public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(envelope.getRoutingKey()); System.out.println(envelope.getExchange()); System.out.println(envelope.getDeliveryTag()); System.out.println(new String(body, "utf-8")); } }; //监听消息 channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer); } }
标签：实现 cpu width mic att 服务器 prope ring bin
原文地址：https://www.cnblogs.com/binwenhome/p/12966193.html
