发表于 | 更新于 | 分类于 | 阅读次数：

摘要:
Netty 中的所有 IO 操作都是异步的，也就是说任何 IO 调用都将立即返回，返回的 ChannelFuture 实例会在操作完成时，通知用户该操作的状态、结果。

阅读全文 »

发表于 | 更新于 | 分类于 | 阅读次数：

字节序

字节序（Byte Order），指的是多字节数据类型，在内存中存储的顺序。
计算机系统，采用字节作为逻辑寻址单位。在处理多字节数据类型时，就要区分字节序。

Big-Endian、Little-Endian

  • Big-Endian 大端（高位字节在前）

低位字节存放在内存的高地址端，高位字节存放在内存的低地址端。
低地址存放最高有效位（MSB）

  • Little-Endian 小端（低位字节在前）

低位字节存放在内存的低地址端，高位字节存放在内存的高地址端。
低地址存放最低有效位（LSB）

说明

  • 大端小端的概念，是面向多字节数据类型的内存存储方式定义的，比如2字节、4字节、8字节的整型、长整型、浮点型等，单字节没有字节序的问题。
  • “前” 是指靠近内存低地址（起始地址），存储在硬盘上就是先写那个字节。
  • 内存地址生长方向为: 从前(左)到后(右)，由低地址到高地址 (这是不变的)。
  • 大端直观，因为与现实生活中，数字书写形式一致: 从前(左)到后(右)，由高位到低位。
    小端符合人的思维，低位值小，就应该放在内存地址小的地方，高位值大就应该放在内存地址大的地方。

举个栗子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

以 4 字节整数为例: 0x12345678

Big-Endian

    低地址           高地址
    -------------------->
    +-+-+-+-+-+-+-+-+-+-+
    | 12 | 34 | 56 | 78 |
    +-+-+-+-+-+-+-+-+-+-+

Little-Endian

    低地址           高地址
    -------------------->
    +-+-+-+-+-+-+-+-+-+-+
    | 78 | 56 | 34 | 12 |
    +-+-+-+-+-+-+-+-+-+-+
阅读全文 »

发表于 | 更新于 | 分类于 | 阅读次数：

rocketmq 源码版本: 4.3.1

Broker

消息中转角色，负责接收消息，消息存储，转发消息，一般也称为 Message Server。
JMS 规范中，称为 Provider。

源码解析

Broker 启动

BrokerStartup

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public static void main(String[] args) {
    start(createBrokerController(args));
}

public static BrokerController createBrokerController(String[] args) {
    // ...
    final BrokerController controller = new BrokerController(
        brokerConfig,
        nettyServerConfig,
        nettyClientConfig,
        messageStoreConfig);
    // remember all configs to prevent discard
    controller.getConfiguration().registerConfig(properties);

    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }
    // ...
}

public static BrokerController start(BrokerController controller) {
    // ...
    controller.start();
    // ...
}
阅读全文 »

发表于 | 更新于 | 分类于 | 阅读次数：

rocketmq 源码版本: 4.3.1

Consumer

Consumer Group

  • Consumer 的 groupName 用于把多个 Consumer 组织到一起，提高并发处理能力、可靠性，groupName需要和消息模式 (MessageModel) 配合使用。

  • rocketmq 支持两种消息模式 (MessageModel): CLUSTERING、BROADCASTING。

    • 在 CLUSTERING 模式下，同一个 consumerGroup (groupName 相同) 里的每个 Consumer 实例 (进程)，只消费所订阅消息的一部分内容，同一个 Consumer Group 里所有的 Consumer 实例消费的内容合起来才是所订阅 Topic 内容的整体，从而达到负载均衡的目的。
    • 在 BROADCASTING 模式下，同一个 Consumer Group 里的每个 Consumer 都能消费到所订阅 Topic 的全部消息，也就是一个消息会被多次分发，被多个 Consumer 消费，也即 BROADCASTING 模式下，Consumer Group 无意义。
阅读全文 »

发表于 | 更新于 | 分类于 | 阅读次数：

rocketmq 源码版本: 4.3.1

示例: 同步发送消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
/**
 * Send Messages Synchronously
 * Reliable synchronous transmission is used in extensive scenes,
 * such as important notification messages, SMS notification, SMS marketing system, etc..
 */
public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException {

        // Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("producer_group_huohu_rocketmq_test");
        // Specify name server addresses.
        producer.setNamesrvAddr("namesrv_address_1;namesrv_address_2");
        // rocketmq-client 默认开启了 vip 通道，vip 通道端口为 10911-2=10909
        // 若 rocketmq-broker 未开启 vip 通道，则报 connect to <......:10909> failed
        producer.setVipChannelEnabled(false);
        producer.setInstanceName("producer_instance_" + System.currentTimeMillis());

        // Launch the instance.
        producer.start();

        for (int i = 0; i < 16; i++) {
            try {
                // Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("topic_huohu_test_simple_message_syn",
                        "tag_a",
                        "key_order_888888_" + i,
                        ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Call send message to deliver message to one of brokers.
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);

                // Send Messages in One-way Mode
                // One-way transmission is used for cases requiring moderate reliability, such as log collection.
                // producer.sendOneway(msg);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        Thread.sleep(1 * 60 * 1000L);

        producer.shutdown();
    }
}
阅读全文 »