<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.0.0</version>
</dependency>

/**
 * Declare a queue
 * @see com.rabbitmq.client.AMQP.Queue.Declare
 * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
 * @param queue the name of the queue
 * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
 * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
 * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
 * @param arguments other properties (construction arguments) for the queue
 * @return a declaration-confirm method to indicate the queue was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                             Map<String, Object> arguments) throws IOException;

/**
 * Publish a message.
 *
 * Publishing to a non-existent exchange will result in a channel-level
 * protocol exception, which closes the channel.
 *
 * Invocations of <code>Channel#basicPublish</code> will eventually block if a
 * <a href="http://www.rabbitmq.com/alarms.html">resource-driven alarm</a> is in effect.
 *
 * @see com.rabbitmq.client.AMQP.Basic.Publish
 * @see <a href="http://www.rabbitmq.com/alarms.html">Resource-driven alarms</a>
 * @param exchange the exchange to publish the message to
 * @param routingKey the routing key
 * @param props other properties for the message - routing headers etc
 * @param body the message body
 * @throws java.io.IOException if an error is encountered
 */
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;

/**
 * No-op implementation of {@link Consumer#handleDelivery}.
 */
@Override
public void handleDelivery(String consumerTag,
                           Envelope envelope,
                           AMQP.BasicProperties properties,
                           byte[] body)
    throws IOException
{
        // no work to do
}

/**
 * Start a non-nolocal, non-exclusive consumer, with
 * a server-generated consumerTag.
 * @param queue the name of the queue
 * @param autoAck true if the server should consider messages
 * acknowledged once delivered; false if the server should expect
 * explicit acknowledgements
 * @param callback an interface to the consumer object
 * @return the consumerTag generated by the server
 * @throws java.io.IOException if an error is encountered
 * @see com.rabbitmq.client.AMQP.Basic.Consume
 * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
 * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
 */
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

package com.lyf.springboot02.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

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

/**
 * @author lyf
 * @date 2019/7/28 18:12
 */
public class Sender {
    public static final String QUEUE_NAME = "simple";

    public static void main(String[] args) throws IOException, TimeoutException {
        /**
         * 1. 获取连接connection
         * 2. 声明消息队列queue name
         * 3. 发送消息message
         * 4. 关闭连接connection
         */
        // 定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setPort(5672);
        factory.setHost("192.168.37.200");
        factory.setUsername("lyf");
        factory.setPassword("123456");
        factory.setVirtualHost("/lyf");
                
        // 创建连接
        Connection connection = factory.newConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        /*  声明一个 Queue
            第一个参数叫 queue 名称
            第二个参数 durable：是否持久化
            第三个参数： Exclusive：排他队列，如果一个队列被声明为排他队列，该队列仅
            对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点：
            其一，排他队列是基于连接可见的，同一连接的不同通道是可以同时访问同一个
            连接创建的排他队列的。
            其二，“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建
            立同名的排他队列的，这个与普通队列不同。
            其三，即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会
            被自动删除的。
            这种队列适用于只限于一个客户端发送读取消息的应用场景。
            第四个参数：Auto-delete:自动删除，如果该队列没有任何订阅的消费者的话，该
            队列会被自动删除。这种队列适用于临时队列。
        */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 发送消息
        String msg = "hello mq";
        // 发送消息到 mq 队列中
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes("utf-8"));
        // 关闭通道
        channel.close();
        // 关闭连接
        connection.close();
    }
}

package com.lyf.springboot02.mq;

import com.rabbitmq.client.*;

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

public class Receiver {
    public static final String QUEUE_NAME = "simple";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setPort(5672);
        factory.setHost("192.168.37.200");
        factory.setUsername("lyf");
        factory.setPassword("123456");
        factory.setVirtualHost("/lyf");
        // 创建连接对象
        Connection connection = factory.newConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 指定队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 获取消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 获取消息并在控制台打印
                String msg = new String(body, "utf-8");
                System.out.println("get msg:" + msg);
            }
        };
        // 监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

channel.basicQos(1);

// 获取消息
Consumer consumer=new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        String msg=new String(body,"utf-8");
        System.out.println("1 号消费者消费消息:msg-->"+msg);
        try {
            Thread.sleep(250);
            channel.basicAck(envelope.getDeliveryTag(), false);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
};

channel.basicConsume(QUEUE_NAME, false, consumer);

/**
 * Actively declare a non-autodelete, non-durable exchange with no extra arguments
 * @see com.rabbitmq.client.AMQP.Exchange.Declare
 * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
 * @param exchange the name of the exchange
 * @param type the exchange type
 * @return a declaration-confirm method to indicate the exchange was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type) throws IOException;

/**
 * Bind a queue to an exchange, with no extra arguments.
 * @see com.rabbitmq.client.AMQP.Queue.Bind
 * @see com.rabbitmq.client.AMQP.Queue.BindOk
 * @param queue the name of the queue
 * @param exchange the name of the exchange
 * @param routingKey the routing key to use for the binding
 * @return a binding-confirm method if the binding was successfully created
 * @throws java.io.IOException if an error is encountered
 */
Queue.BindOk queueBind(String queue, String exchange, String routingKey) throws IOException;

// 定义交换机名
final String EXCHANGE_NAME="hello-fanout";
// 声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
// 发布消息
String msg = "Hello";
channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes("utf-8"));

// 定义交换机名
final String EXCHANGE_NAME="hello-fanout";
// 声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
// 创建默认排他队列
String queueName = channel.queueDeclare().getQueue();
// 绑定队列到路由
channel.queueBind(queueName,EXCHANGE_NAME,"");

// 发布消息
String msg = "Hello-info";
// 设置路由key
String routingKey = "info";
// 绑定队列到路由
channel.basicPublish(EXCHANGE_NAME,routingKey,null,msg.getBytes("utf-8"));

// 设置路由key
String routingKey = "info";
// 绑定队列到路由
channel.queueBind(queueName,EXCHANGE_NAME,routingKey);

// 声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

// 开启确认模式
channel.confirmSelect();
...
// 等待确认
if(channel.waitForConfirms()){
    System.out.println("发送成功。。。");
}

// 开启确认模式
channel.confirmSelect();
...
// 等待确认
channel.waitForConfirmsOrDie();
System.out.println("发送成功。。。");

// 定义集合存放seqNO
final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());
...
// 开启确认模式
channel.confirmSelect();

// 绑定监听器
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        if(multiple){
            System.out.println("mutil-succe: "+deliveryTag);
            confirmSet.headSet(deliveryTag+1L).clear();
        }else{
            System.out.println("single-succe: "+deliveryTag);
            confirmSet.remove(deliveryTag);
        }
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        if(multiple){
            System.out.println("mutil-fail: "+deliveryTag);
            confirmSet.headSet(deliveryTag+1L).clear();
        }else{
            System.out.println("single-fail: "+deliveryTag);
            confirmSet.remove(deliveryTag);
        }
    }
});
...
// 获取编号
long seqNo = channel.getNextPublishSeqNo();
channel.basicPublish(EXCHANGE_NAME,routingKey,null,msg.getBytes("utf-8"));
System.out.println("Send: "+msg);
//添加编号
confirmSet.add(seqNo);