pull类型消息中间件-消息发布者(一)

消息集群架构

对于发送方来说的关键几要素

topic

消息的主题，由用户定义。类似于知乎的话题，Producer发送消息的时候需要指定发送到某一个topic下面，Consumer从某一个topic下面消费消息。

每次发送一条消息的时候，给消息加一个Tag，方便Consumer过滤消息

message

消息，负载发送的消息的信息。在生产者，服务端和消费者之间传输

queue

queue就是metaq中具体用来存数消息的数据结构，每一个topic下面对应多个queue，以目录的形式分开存储在磁盘上。我们在最初的网络结构中就描述了brokerserver集群，然而对应到具体某一个brokerserver的存储，就是queue+commitlog

之前介绍brokerserver的时候有简单提过DefaultMessageStore类，这是metaq消息存储的默认实现类，里面存储的具体配置，也包含了commitlog，consumequeue，关于消息加载，刷盘，恢复，清理等存储设计都在这里。

Group

消费者可以是多个消费者共同消费一个 topic 下的消息,每个消费者消费部分消息。这些消费者就组成一个分组,拥有同一个分组名称,通常也称为消费者集群。

集群消费，一条消息只会被同一个group里一个消费者消费。不同group之间相互不影响。

广播消费，一条消息会被同一个group里每一个消费端消费

消息发送者实例化

Producer 消息生产者，负责产生消息，一般由业务系统负责产生消息。

一般一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例。

消息子类型用 tags 来标识，tags 可以由应用*设置。只有发送消息设置了 tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。

我们可以看出start()方法主要做这么几件事：

检查配置是否正确,checkConfig();
注册produce,registerProducer();
mQClientFactory.start();
向所有的broker发送的Heartbeat。

发送方的核心服务：

 //Start request-response channel

 this.mQClientAPIImpl.start();

 //Start various schedule tasks

 this.startScheduledTask();

 //Start pull service

 this.pullMessageService.start();

 //Start rebalance service

 this.rebalanceService.start();

 //Start push service

 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

消息发送流程

pull类型消息中间件-消息发布者(一)

metaQ的底层通信还是基于netty的。

消息载体

private String topic;

private int flag;

private Map<String, String> properties;

private byte[] body;

private int queueId;

private int storeSize;

private long queueOffset;

private int sysFlag;

private long bornTimestamp;

private SocketAddress bornHost;

private long storeTimestamp;

private SocketAddress storeHost;

private String msgId;

private long commitLogOffset;

private int bodyCRC;

private int reconsumeTimes;

private long preparedTransactionOffset;

这里需要强调的是，Message是为了便于metaq逻辑操作而定义的偏向于业务逻辑的类，实际上的网络传输类是RemotingCommand，结构如下

pull类型消息中间件-消息发布者(一)

 public ByteBuffer encode() {

        int length = 4;

        byte[] headerData = this.buildHeader();

        length += headerData.length;

        if (this.body != null) {

            length += body.length;

        }

        ByteBuffer result = ByteBuffer.allocate(4 + length);

        result.putInt(length);

        result.putInt(headerData.length);

        result.put(headerData);

        if (this.body != null) {

            result.put(this.body);

        }

        result.flip();

        return result;

    }

producer的send()方法：

pull类型消息中间件-消息发布者(一)

客户端从zookeeper上获取publish的topic对应的broker和分区列表（按照brokerId和partition的顺序排列组织成一个有序的分区列表），生产者在发送消息的时候必须选择一个分区来发送消息，发送的时候按照从头到尾循环往复的方式选择来发送消息。

默认调用的是同步发送方法。

调用netty中的 com.alibaba.rocketmq.remoting.netty.NettyRemotingClient#invokeSync

  public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)

            throws InterruptedException, RemotingConnectException, RemotingSendRequestException,

            RemotingTimeoutException {

        final Channel channel = this.getAndCreateChannel(addr);

        if (channel != null && channel.isActive()) {

            try {

                if (this.rpcHook != null) {

                    this.rpcHook.doBeforeRequest(addr, request);

                }

                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis);

                if (this.rpcHook != null) {

                    this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel),

                        request, response);

                }

                return response;

            }

            catch (RemotingSendRequestException e) {

                log.warn("invokeSync: send request exception, so close the channel[{}]", addr);

                this.closeChannel(addr, channel);

                throw e;

            }

            catch (RemotingTimeoutException e) {

                log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);

                throw e;

            }

        }

        else {

            this.closeChannel(addr, channel);

            throw new RemotingConnectException(addr);

        }

    }

同时可以设置调用异步发送方法。

调用com.alibaba.rocketmq.remoting.netty.NettyRemotingClient#invokeAsync

 public void invokeAsync(String addr, RemotingCommand request, long timeoutMillis,

            InvokeCallback invokeCallback) throws InterruptedException, RemotingConnectException,

            RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {

        final Channel channel = this.getAndCreateChannel(addr);

        if (channel != null && channel.isActive()) {

            try {

                if (this.rpcHook != null) {

                    this.rpcHook.doBeforeRequest(addr, request);

                }

                this.invokeAsyncImpl(channel, request, timeoutMillis, invokeCallback);

            }

            catch (RemotingSendRequestException e) {

                log.warn("invokeAsync: send request exception, so close the channel[{}]", addr);

                this.closeChannel(addr, channel);

                throw e;

            }

        }

        else {

            this.closeChannel(addr, channel);

            throw new RemotingConnectException(addr);

        }

    }

这两个方法一个为同步，一个为异步。而同步与异步的主要实现则来自于ResponseFuture类。通过countDownLatch的使用，完成了同步操作（详情参看此类的putResponse方法及waitResponse方法）；通过传递invokeCallback完成异步操作

秒客网

pull类型消息中间件-消息发布者(一)

消息集群架构

对于发送方来说的关键几要素

消息发送者实例化

消息发送流程

消息载体

相关文章