【RabbitMq源码阅读】分析RabbitMq发送消息源码

一：基本介绍

本文通过demo构建测试代码，debug分析的方法查看RabbitMq源码。

rabbit的中文文档：官方中文文档

二：测试Demo

2.1 引入Springboot整合的RabbitMq依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.2 手写获取RabbitMq的连接，通道等信息

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * RabbitMq Source Test
 * @author c
 * date: 2024-9-26 19:12:27
 */
public class RabbitMqSourceTest {

    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 设置rabbitmq的服务器地址
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setPort(AMQP.PROTOCOL.PORT);
        // 建立连接
        Connection conn = connectionFactory.newConnection();

        String exchange = "test-Exchange";
        String queueName = "test-Queue";
        String key = "test-Exchange-key";
        String msg = "测试消息";

        // 创建一个channel
        Channel channel = conn.createChannel();

        // 创建一个直连交换机
        channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT);

        // 创建一个队列
        channel.queueDeclare(queueName, true, false, false, null);

        // 绑定队列
        channel.queueBind(queueName, exchange, key);

        // 发送消息
        channel.basicPublish(exchange, key, null, msg.getBytes());

        channel.close();
        conn.close();   
    }

}

上面的基本流程

简单理解为：通过连接，获取通道，数据传输。

基本步骤：

获取连接(Connection)
获取通道(channel)
创建交换机(Exchange)
创建队列(Queue)
队列通过key绑定交换机(Bind)
往交换机中的key发送消息
其他方法

(上面创建消息的也是可以直接创建队列，进行消息的发送，源码中会将没有创建exchange设置成默认的，具体可以自己查看一下)

三：详细分析步骤

3.0 获取连接(Connection)

 public Connection newConnection(ExecutorService executor, AddressResolver addressResolver, String clientProvidedName)
        throws IOException, TimeoutException {
        if(this.metricsCollector == null) {
            this.metricsCollector = new NoOpMetricsCollector();
        }
        // make sure we respect the provided thread factory
        FrameHandlerFactory fhFactory = createFrameHandlerFactory();
        ConnectionParams params = params(executor);
        // set client-provided via a client property
        if (clientProvidedName != null) {
            Map<String, Object> properties = new HashMap<String, Object>(params.getClientProperties());
            properties.put("connection_name", clientProvidedName);
            params.setClientProperties(properties);
        }
        // 如果设置 自动发送为 true
        if (isAutomaticRecoveryEnabled()) {
            // see 
         
  com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory#newConnection
            // 创建连接            
AutorecoveringConnection conn = new AutorecoveringConnection(params, fhFactory, addressResolver, metricsCollector);
            // 初始化
            conn.init();
            return conn;
        } else {
            // 通过 addrs  获取
            List<Address> addrs = addressResolver.getAddresses();
            Exception lastException = null;
            for (Address addr : addrs) {
                try {
                    // 创建对应的 FrameHandler 
                    FrameHandler handler = fhFactory.create(addr, clientProvidedName);
                    // 创建连接                    
                    AMQConnection conn = createConnection(params, handler, metricsCollector);    
                    conn.start();
                    this.metricsCollector.newConnection(conn);
                    return conn;
                } catch (IOException e) {
                    lastException = e;
                } catch (TimeoutException te) {
                    lastException = te;
                }
            }
            if (lastException != null) {
                if (lastException instanceof IOException) {
                    throw (IOException) lastException;
                } else if (lastException instanceof TimeoutException) {
                    throw (TimeoutException) lastException;
                }
            }
            throw new IOException("failed to connect");
        }
    }

3.1 创建渠道(Channel)

    @Override
    public Channel createChannel() throws IOException {
        // 确定开启状态
        ensureIsOpen();
        ChannelManager cm = _channelManager;
        if (cm == null) return null;
        // 重点可以看下这里: 创建channel
        Channel channel = cm.createChannel(this);
        // 通过 metricsCollector 创建新的channel
        metricsCollector.newChannel(channel);
        return channel;
    }

createChannel 方法：

这里主要是通过channelNumberAllocator分配到一个channelNumber，可以理解为一个唯一标识，具体可以自行看一下它的实现

    public ChannelN createChannel(AMQConnection connection) throws IOException {
        ChannelN ch;
        synchronized (this.monitor) {
            // 通过 channelNumberAllocator 获取到一个 channelNumber 
            int channelNumber = channelNumberAllocator.allocate();
            if (channelNumber == -1) {
                return null;
            } else {
                ch = addNewChannel(connection, channelNumber);
            }
        }
        ch.open(); // now that it's been safely added
        return ch;
    }

3.2 创建交换机(Exchange)

    public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException {
        final AMQP.Exchange.DeclareOk ok = delegate.exchangeDeclare(exchange, type, durable, autoDelete, internal, arguments);
        RecordedExchange x = new RecordedExchange(this, exchange).
          type(type).
          durable(durable).
          autoDelete(autoDelete).
          arguments(arguments);
        // 记录当前的交换机
        recordExchange(exchange, x);
        return ok;
    }

进入delegate.exchangeDeclare方法，可以看到控制台会创建成功exchange：

3.3 创建队列(Queue)

    @Override
    public AMQP.Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException {
        // 这里执行完成会创建成功队列
        final AMQP.Queue.DeclareOk ok = delegate.queueDeclare(queue, durable, exclusive, autoDelete, arguments);
        RecordedQueue q = new RecordedQueue(this, ok.getQueue()).
            durable(durable).
            exclusive(exclusive).
            autoDelete(autoDelete).
            arguments(arguments);
        if (queue.equals(RecordedQueue.EMPTY_STRING)) {
            q.serverNamed(true);
        }
        recordQueue(ok, q);
        return ok;
    }

    @Override
    public Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive,
                                        boolean autoDelete, Map<String, Object> arguments)
        throws IOException
    {
        validateQueueNameLength(queue);
        return (Queue.DeclareOk)
               // 通过rpc申明 信息
               exnWrappingRpc(new Queue.Declare.Builder()
                               .queue(queue)
                               .durable(durable)
                               .exclusive(exclusive)
                               .autoDelete(autoDelete)
                               .arguments(arguments)
                              .build())
               .getMethod();
    }

可以看到此时虽然创建了queue，但是并未绑定到exchang上面，需要进行下面的绑定

3.4 绑定队列

    @Override
    public AMQP.Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException {
        // 绑定队列
        AMQP.Queue.BindOk ok = delegate.queueBind(queue, exchange, routingKey, arguments);
        recordQueueBinding(queue, exchange, routingKey, arguments);
        return ok;
    }

    @Override
    public Queue.BindOk queueBind(String queue, String exchange,
                                  String routingKey, Map<String, Object> arguments)
        throws IOException
    {
        validateQueueNameLength(queue);
        return (Queue.BindOk)    
               // 通过 rpc 申明绑定信息
               exnWrappingRpc(new Queue.Bind.Builder()
                               .queue(queue)
                               .exchange(exchange)
                               .routingKey(routingKey)
                               .arguments(arguments)
                              .build())
               .getMethod();
    }

3.5 发送消息

    @Override
    public void basicPublish(String exchange, String routingKey,
                             boolean mandatory, boolean immediate,
                             BasicProperties props, byte[] body)
        throws IOException
    {
        if (nextPublishSeqNo > 0) {
            unconfirmedSet.add(getNextPublishSeqNo());
            nextPublishSeqNo++;
        }
        if (props == null) {
            props = MessageProperties.MINIMAL_BASIC;
        }
        // 组装消息
        AMQCommand command = new AMQCommand(
            new Basic.Publish.Builder()
                .exchange(exchange)
                .routingKey(routingKey)
                .mandatory(mandatory)
                .immediate(immediate)
                .build(), props, body);
        try {
            // 发送消息
            transmit(command);
        } catch (IOException e) {
            metricsCollector.basicPublishFailure(this, e);
            throw e;
        }
        // 推送当前的channel 进行发布
        metricsCollector.basicPublish(this);
    }