/**
 * 通配符模式:发送消息
 */
public class Producer {
    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchage";
    //队列名称
    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 Exception {
        //1. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
        channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
        channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);

        //5. 队列绑定到交换机上
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");

        //5. 队列绑定到交换机上
        channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
        channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
        //6. 发送消息；
        String message = "商品新增。通配符模式 ；routing key 为 item.insert ";

        /**
         * 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
         * 参数2:路由key,简单模式中可以使用队列名称
         * 参数3:消息其它属性
         * 参数4:消息内容
         */
        channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        message = "商品修改。通配符模式 ；routing key 为 item.update ";

        /**
         * 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
         * 参数2:路由key,简单模式中可以使用队列名称
         * 参数3:消息其它属性
         * 参数4:消息内容
         */
        channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        message = "商品删除。通配符模式 ；routing key 为 item.delete ";

        /**
         * 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
         * 参数2:路由key,简单模式中可以使用队列名称
         * 参数3:消息其它属性
         * 参数4:消息内容
         */
        channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

/**
 * 通配符模式；消费者接收消息
 */
public class Consumer1 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
    
        //4. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
            }
        };
        //6. 监听队列
        /**
         * 参数1:队列名
         * 参数2:是否要自动确认；设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除；
         * 如果设置为false则需要手动确认
         * 参数3:消费者
         */
        channel.basicConsume(Producer.TOPIC_QUEUE_1, true, defaultConsumer);
    }
}