RabbitMQ系列十二，通过rabbitClient 发送和消

消息的生产和消费，非确认模式

生产端代码

package com.edu.mq.rabbitmq;

import com.rabbitmq.client.*;

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

/**
 * 类说明：
 *
 * @author zhangkewei
 * @date 2018/11/21下午7:34
 */
public class Send1 {

    public static void main(String[] args) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
        //通过uri方式创建工厂，和之前创建工厂的方式不一样
        factory.setUri("amqp://guest:guest@localhost:5672");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();

        String exchange = "email";
        String queue = "sms_forget_queue";

        /**
         * 设置消息的属性,持久化，并且UTF-8编码
         */
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8").build();

        /**
         * 创建queue
         */
        channel.queueDeclare(queue, true, false, false, null);
        /**
         * 不指定exchange，就是使用当前virtual host下面的defaout exchange ，不需要绑定，并将消息发送到和routingKey名字相同的队列里
         * */
        channel.basicPublish("", queue,props,"通过默认exchange发送消息".getBytes());

        /**
         * 发送消息到指定exchange中，需要先进行绑定，如果不先进行绑定，消费发送后会被丢弃。
         * 消息的发送放，只关心exchange和绑定，但是绑定一般需要绑定知道exchange，和quque及绑定关系，所以，
         * 一般exchange和queue及绑定关系由生产端确定，但这并不是一定的，消费端同样可以创建好队列及绑定关系，
         * 毕竟rabbit支持重复创建，即使已经存在也不会对原有的exchange，queue，binding有影响。
         */
        channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT, true);
        channel.queueBind(queue,exchange,"log_routingKey");
        channel.basicPublish(exchange,"log_routingKey",props,"指定exchange绑定之后，发送消息".getBytes());


        TimeUnit.SECONDS.sleep(20);

        channel.close();
        conn.close();
    }
}

消费端代码

package com.edu.mq.rabbitmq;

import com.rabbitmq.client.*;

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

/**
 * 类说明：消费者
 *
 * @author zhangkewei
 * @date 2018/11/21下午7:35
 */
public class Consumer1 {
    public static void main(String[] args) throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
        //通过uri方式创建工厂，和之前创建工厂的方式不一样
        factory.setUri("amqp://guest:guest@localhost:5672");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();

        /**
         * 消费者只关心需要消费的队列
         */
        String queue = "sms_forget_queue";
        /**
         * 通过内部类的方式，继承一个DefaultConsumer类的一个子类，并重写handleDelivery方法
         * 其中第一个参数表示队列，第二个参数表示是否自动确认，第三个参数具体的消费该队列中的消息
         */

        channel.basicConsume(queue, true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException {

                System.out.println("收到消息：consumerTag = "+ consumerTag+"consumerTag = " +
                        envelope+" consumerTag = " + properties+" 消息内容为： " + new String(body));

            }
        });
        TimeUnit.SECONDS.sleep(20);

        channel.close();
        conn.close();
    }
}

执行过程中，管理界面消息的消费和接收情况

说明：
1，上面的消息的生产和消息的消费没有得到确认，即，消息生产端，没有保证消息一定发送到了broke，消息的消费端，也没有确保消息消费过程中，如果消费失败了，mq会再次发送消息。
2，生产者和消费者启动的先后，对消息的消费没有影响。但是在实际的开发过程中，一般先启动消费者，在启动生产者，避免消息的大量积压，