首先我们还是先看一下Rabbitmq的工作原理图

         从图上我们可以看到，无论是生产者还是消费者我们都需要进行connection并且获取相应的channel，所以为了方便，建议把这部分操作抽取成一个工具类RabbitMqUtils。

工具类RabbitMqUtils

public class RabbitMqUtils {
    public static Channel getChannel() throws IOException, TimeoutException {

        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 工厂ip 连接rabbitMQ的队列 你的ip  通过ipconfig查看ipv4
        factory.setHost("192.168.50.12");
        //用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");

        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        return channel;
    }
}

生产者Producer

public class Producer {
    // 队列名称
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 调用连接并获取通道工具类
        Channel channel = RabbitMqUtils.getChannel();

        /**
           * 生成队列
           *1.队列名称
           *2.队列里面的消息是否进行持久化（磁盘） 默认情况消息存储在内存中
           *3.该队列是否只提供一个消费者进行消费 是否进行消息的共享 true可以多个消费者消费
           *4.是否自动删除 最后一个消费者开连接以后  该队列是否自动删除 true自动删除
           *5.其他参数 延迟 死信等
           */

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        // 发消息
        String message = "hello,world";

        /**
         * 发送一个消息
         * 1.发送到哪个交换机
         * 2.路由的key值是哪一个 本次是队列名称
         * 3.其他参数信息
         * 4.发送消息的消息体
         *
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

        System.out.println("消息发送完毕");

        // 控制台输出消息发送完毕

 后台已经有一个就绪消息

 消费者Consumer 

public class Consumer {
    // 队列的名称
    public static final String QUEUE_NAME = "hello";

    // 接收消息
    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();

        // 推送的消息如何进行消费的接口回调
        DeliverCallback deliverCallback = (consumerTag,message)->{
            System.out.println(new String (message.getBody()));
        };
      // 取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = consumerTag->{
            System.out.println("消息消费被中断");
        };

         /**
         * 1.消费的目标队列
         * 2.消费成功之后是否要自动应答 true自动  false 手动
         * 3.消费未成功消费的回调
         * 4.消费者取消消费的回调
         */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);

    }

}
    // 控制台输出 hello,world

 这个时候后台可以看到消息已经被消费了，还可以看到消费时间

来看一下 channel.basicConsume()方法

 

 前两个参数注释已经很清晰，我上面已经在代码上注释了，来看一下第三个和第四个参数

首先可以看到他的注解@FunctionalInterface，所以这是一个函数式接口，所以在我们获取这个参数的时候可以使用lambda表达式即函数式写法。