RabbitMQ的入门篇-RabbitMQ-HelloWorld

本文是RabbitMQ的入门篇

安装篇:https://blog.csdn.net/qq_58267473/article/details/124600014

RabbitMQ 是一个消息中间件：它接受并转发消息。你可以把它当做公司，客户把需求提给公司,项目经理项目经理会向下分配工作，按照这种逻辑 RabbitMQ 是一个公司，一个项目经理分配工作。RabbitMQ 与公司的主要区别在于，它不处理需求而是接收，
存储和转发消息数据

生产者
产生数据发送消息的程序是生产者
交换机
交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息
推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推
送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
队列
队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存
储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可
以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
消费者
消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费
者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

介绍完了我们来看下Rabbit的核心部分

本篇文章对第一部分进行学习

    
        
            org.apache.maven.plugins
            maven-compiler-plugin
            
                8
                8
            
        
    


    
    
        com.rabbitmq
        amqp-client
        5.8.0
    
    
    
        commons-io
        commons-io
        2.6
    

package com.win.rabbitmq.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;


public class Producer {
    public static final String Queue_NAME="hello";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 创建一个连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //工厂IP,连接Rabbit的对列
        connectionFactory.setHost("192.168.159.105");
        //用户名
        connectionFactory.setUsername("admin");
        //密码
        connectionFactory.setPassword("ok");
        //创建连接
        Connection con = connectionFactory.newConnection();
        //获取信道
        Channel channel = con.createChannel();
        
        channel.queueDeclare(Queue_NAME,false,false,false,null);
        //发送的消息
        String msg="Hello World";
        
        channel.basicPublish("", Queue_NAME, null, msg.getBytes());
        System.out.println("消息发生完毕");

    }
}

package com.win.rabbitmq.one;

import com.rabbitmq.client.*;

import java.awt.geom.FlatteningPathIterator;
import java.io.IOException;
import java.util.MissingFormatArgumentException;
import java.util.concurrent.TimeoutException;


public class Consumer {
    public static final String QUEUE_NAME="hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.159.105");
        factory.setUsername("admin");
        factory.setPassword("ok");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //消费者未成功消费的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String s = new String(message.getBody());
            System.out.println(consumerTag+s);
        };
        CancelCallback cancelCallback = consumerTag -> System.out.println("消息消费被中断");

        
        channel.basicConsume(QUEUE_NAME, true,deliverCallback,cancelCallback);
    }
}

先启动生产者代码再启动消费者代码,原因是生产者没启动时并没有名为hello的队列

因为获取信道的代码是重复的,所以我们可以把它抽出来

package com.win.rabbitmq.utils;

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

import com.rabbitmq.client.Channel;


public class RabbitUtils {
    public static Channel getChannel() throws Exception {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.159.105");
        factory.setUsername("admin");
        factory.setPassword("ok");
        Connection connection = factory.newConnection();
        return connection.createChannel();
    }
}