生产者
package com.qf.mq2302.topic;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;public class Pubisher {public static final String EXCHANGE_NAME="mypubilisher";public static void main(String[] args) throws Exception {Connection connection = MQUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,"topic");String msg="好好学习";String routingkey="lazy.orange.rabbit";channel.basicPublish(EXCHANGE_NAME,routingkey,null,msg.getBytes("utf-8"));channel.close();connection.close();}
}
消费者1
package com.qf.mq2302.topic;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;import java.io.IOException;public class MyConsumer01 {public static final String EXCHANGE_NAME="mypubilisher";public static void main(String[] args) throws Exception {Connection connection = MQUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,"topic");String queue = channel.queueDeclare().getQueue();channel.basicQos(1);//绑定队列和交换机String routingkey="*.orange.*";channel.queueBind(queue,EXCHANGE_NAME,routingkey);channel.basicConsume(queue, false, new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {byte[] body = message.getBody();String s = new String(body, "utf-8");System.out.println(s);channel.basicAck(message.getEnvelope().getDeliveryTag(),false);}},consumerTag -> {});}}
消费者2
package com.qf.mq2302.topic;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;import java.io.IOException;public class MyConsumer02 {public static final String EXCHANGE_NAME="mypubilisher";public static void main(String[] args) throws Exception {Connection connection = MQUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,"topic");String queue = channel.queueDeclare().getQueue();channel.basicQos(1);//绑定队列和交换机String routingkey2="*.*.rabbit";String routingkey3="lazy.#";channel.queueBind(queue,EXCHANGE_NAME,routingkey3);channel.queueBind(queue,EXCHANGE_NAME,routingkey2);channel.basicConsume(queue, false, new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {byte[] body = message.getBody();String s = new String(body, "utf-8");System.out.println(s);channel.basicAck(message.getEnvelope().getDeliveryTag(),false);}},consumerTag -> {});}}
