[大数据]Java操作RabbitMQ单一生产-消费者模式

1 介绍RabbitMQ的图形界面

2 操作前配置

2.1 添加用户

2.2 添加Virtual host

3 代码编写

3.1 生产-消费者模式

/**
 * @author 17122
 * 生产者消费者模式
 */
public class One2One {

    public static void producer() throws IOException, TimeoutException {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //链接地址 服务器IP地址
        connectionFactory.setHost("127.0.0.1");
        //端口号
        connectionFactory.setPort(5672);
        //登录名
        connectionFactory.setUsername("ymx");
        //密码
        connectionFactory.setPassword("123");
        //路径
        connectionFactory.setVirtualHost("/one2one");
        //新建链接
        Connection connection = connectionFactory.newConnection();
        //创建通道
        Channel channel = connection.createChannel();
        /**
         * '参数1':用来声明通道对应的队列
         * '参数2':用来指定是否持久化队列
         * '参数3':用来指定是否独占队列
         * '参数4':用来指定是否自动删除队列
         * '参数5':对队列的额外配置
         */
        channel.queueDeclare("hello", true, false, false, null);
        /**
         * '参数1':指定交换机
         * '参数2':指定routingKey（队列）
         * '参数3':其他配置
         * '参数4':发布的信息的字节数组
         */
        channel.basicPublish("", "hello", null, "hello rabbitmq".getBytes());
        channel.close();
        connection.close();
    }

    public static void consumer() throws IOException, TimeoutException {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //链接地址 服务器IP地址
        connectionFactory.setHost("127.0.0.1");
        //端口号
        connectionFactory.setPort(5672);
        //登录名
        connectionFactory.setUsername("ymx");
        //密码
        connectionFactory.setPassword("123");
        //路径(虚拟机)
        connectionFactory.setVirtualHost("/one2one");
        //新建链接
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        /**
         * '参数1':用来声明通道对应的队列
         * '参数2':用来指定是否持久化队列
         * '参数3':用来指定是否独占队列
         * '参数4':用来指定是否自动删除队列
         * '参数5':对队列的额外配置
         */
        channel.queueDeclare("hello", true, false, false, null);
        /**
         * '参数1':指定对应的队列
         * '参数2':如果服务器在发送后考虑确认消息，则为True; 如果服务器期望明确的确认，则为False  
         * '参数3':消费者
         */
        channel.basicConsume("hello", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body));
            }
        });
    }


    public static void main(String[] args) throws IOException, TimeoutException {
        for (int i = 0; i < 10; i++) {
            producer();
        }
        consumer();
    }
}

测试结果：