7、RabbitMQ-主题模式

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.util.ConnectionUtils;

public class Send {
    
    private static final String EXCHANGE_NAME = "exchange_topic";
    
    public static void main(String[] args) throws IOException, TimeoutException {
        
        Connection conn = ConnectionUtils.getConnection();
        
        Channel channel = conn.createChannel();
        
        //exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        
        String msg = "商品.....";
        
        //绑定路由
        String routingKey = "goods.add";
        channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
        
        channel.close();
        conn.close();
        
    }

}

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.util.ConnectionUtils;

public class Receive {
    
    private static final String QUEUE_NAME="test_topic1";
    private static final String EXCHANGE_NAME = "exchange_topic";
    
    public static void main(String[] args) throws IOException, TimeoutException {
        
        Connection conn = ConnectionUtils.getConnection();
        
        Channel channel = conn.createChannel();
        
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        channel.basicQos(1);
        
        //绑定队列到交换机转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "goods.add");

                //定义一个消费者
                Consumer consumer = new DefaultConsumer(channel){
                    //收到消息就会触发这个方法
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                            throws IOException {
                        String msg = new String(body,"utf-8");
                        System.out.println("消费者1接收到的消息" + msg);
                        
                        try {
                            Thread.sleep(1500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }finally{
                            System.out.println("消费者1处理完成！");
                            //手动回执
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                //监听队列
                //自动应答false
                boolean autoAck = false;
                channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.util.ConnectionUtils;

public class Receive2 {
    
    private static final String QUEUE_NAME="test_topic2";
    private static final String EXCHANGE_NAME = "exchange_topic";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConnection();
        Channel channel = conn.createChannel();
        
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);
        
        //绑定队列到交换机转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "goods.#");
        
                //定义一个消费者
                Consumer consumer = new DefaultConsumer(channel){
                    //收到消息就会触发这个方法
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                            throws IOException {
                        String msg = new String(body,"utf-8");
                        System.out.println("消费者2接收到的消息" + msg);
                        
                        try {
                            Thread.sleep(1500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }finally{
                            System.out.println("消费者2处理完成！");
                            //手动回执
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                //监听队列
                //自动应答false
                boolean autoAck = false;
                channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}