生产者
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() {
@Override
public 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() {
@Override
public 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 -> {});
}
}