rabbitmq入门

安装

1
docker run -d --name web_rabbitmq -p 15673:15672 -p 5673:5672 -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=sdadgz.cn --network website_net rabbitmq:3.9-management

hello world

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Test
void rabbitPublish() throws IOException, TimeoutException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("sdadgz.cn");
    factory.setPort(5673);
    factory.setVirtualHost("/");
    factory.setUsername("root");
    factory.setPassword("sdadgz.cn");

    @Cleanup Connection connection = factory.newConnection();
    @Cleanup Channel channel = connection.createChannel();
    
    channel.queueDeclare("hello_world", true, false, false, null);

    for (int i = 0; i < 100; i++) {
        /*
        	交换机名
        	队列名
        	参数
        	内容
        */
        channel.basicPublish("", "hello_world", null, (i + "massage~~").getBytes(StandardCharsets.UTF_8));
    }
}

消费者

可以多个，还有，别手欠加上close

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Test
void rabbitConsumer() throws IOException, TimeoutException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("sdadgz.cn");
    factory.setPort(5673);
    factory.setVirtualHost("/");
    factory.setUsername("root");
    factory.setPassword("sdadgz.cn");

    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    channel.queueDeclare("hello_world", true, false, false, null);

    channel.basicConsume("hello_world", 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));
        }
    });
}

workqueues

hello world多个消费者就是work queues

pub/sub

1
2
3
4
5
6
7
8
channel.exchangeDeclare([交换机名称], [交换机类型], [持久化], [自动啥暗处], [内部使用，一般false], [参数]);
交换机类型：
	direct定向
	fanout广播
	topic
	headers参数匹配
channel.queueDeclare([队列名], [是否独占], [是否自动删除], [参数]);
channel.queueBind([队列名], [交换机名], [routingKey 广播设空字符串])；

routing