RabbitMq简单模式

定义一个生产者,负责发送消息到队列中

/**
 * @author zjh
 * 生产者发信息
 */
public class Producer {

    /**
     * 队列名称
     */
    public static final String QUEUE_NAME = "hello";

    /**
     * 发消息
     */
    public static void main(String[] args) throws Exception {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 工厂IP 连接RabbitMq的队列
        factory.setHost("localhost");
        // 用户名
        factory.setUsername("zjh");
        // 密码
        factory.setPassword("zjh");

        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        /*
            生成一个队列
            1.队列名称
            2.队列里面的消息是否持久化，默认情况消息存储在内存中
            3.该队列是否只提供一个消费者进行消费 是否进行消息共享，true可以多个消费者消费
            4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true自动删除 false不自动删除
            5.其他参数
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 发消息 初次使用
        String message = "hello world";

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

定义消费者，消费指定队列

/**
 * @author zjh
 * 消费消息
 */
public class Consumer {

    /**
     * 队列名称
     */
    public static final String QUEUE_NAME = "hello";

    /**
     * 接收消息
     */
    public static void main(String[] args) throws Exception {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 工厂IP 连接RabbitMq的队列
        factory.setHost("192.168.10.119");
        // 用户名
        factory.setUsername("zjh");
        // 密码
        factory.setPassword("zjh");

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

        // 声明接收消息回调
        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);
    }
}