RocketMq_随笔_内存溢出

RocketMq 1、RocketMQ的系统架构 1、Producer (生产者)

RocketMQ中的消息生产者都是以生产组（Producer Group）的形式出现的，生产组是同一类生产者的集合。
这类Producer发送相同的Topic类型的消息

2、Consumer(消费者)

RocketMQ中的消息消费者都是以消费组（Consumer Group）的形式出现的，消费组是同一类消费者的集合。
这类Consumer消费的是同一个Topic类型的消息，消费组使得在消息消费方面，
实现**负载均衡**(将一个Topic中的不同的Queue平均分配给同一个Consumer Group 中不同的Consumer)和
**容错**(一个Consumer挂了，该Consumer Group 中的其它Consumer可以接着消费原Consumer消费的Queue)。

3、Name Server(Broker与Topic路由的注册中心，支持Broker的动态注册与发现) （1）Broker管理：接受Broker集群的注册信息并且保存下来作为路由信息的基本数据：提供心跳检测机制，检查Broker是否还存活。（2）路由信息管理：每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer可以获取整个Broker集群的路由信息，从而进行消息的投递和消费（3）路由注册： NameServer通常也是以集群的方式部署，不过NameServer是无状态的，NameServer集群中的各个节点间是无差异的，各节点间相互不进行信息通讯，那各节点中的数据是如何进行数据同步的呢？再Broker节点启动时，轮询与每个NameServer节点建立长连接，发送注册请求，在NameServer内部维护着一个Broker列表，用来动态存储Broker的信息。（NameServer扩容是需要哦配置NameServer相关的信息不然Broker的数据同步不到新的NameServer上）Broker节点会每30秒发送一次心跳至NameServer，心跳包含BrokerId、Broker地址、Broker名称、Broker所属集群名称，NameServer收到心跳包会更新心跳时间戳记录Broker的最新存活时间。（4）路由剔除：当Broker宕机了，NameServer接受Broker心跳时没有接收到，会进行删除掉当前没有接收到心跳的Broker （5）路由发现： 1、Push模型: 当Name Server的Topic路由数据发送变化时会主动推送消息给客户端。2、Pull模型：客户端会每隔30秒向NameServer拉取一次路由数据。3、LongPolling模型：长轮询模型，客户端和服务端保持长连接，比如占用30秒（无论30秒是否有数据更新都会建立连接），等30秒的链接时间到后链接取消。（6）客户端NameServer选择策略: 进行随机取模然后取第一台 4、Broker(Broker充当着消息中转角色，负责存储信息、转发消息。Broker在RocketMq系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求做准备。Broker同时也存储着消息相关的元数据，包括消费组消费进度偏移offset、主题、队列等)。（1）、Remoting Module: 整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则油一下模块构成。（2）、Client Manger: 客户端管理器，负责接收、解析客户端（Producer/Consumer）请求，管理客户端。列如：维护Consumer的Topic订阅信息. （3）、HA Service：高可用服务，提供MasterBroker和SlaveBroker之间的数据同步功能。（部署一个Broker之后会产生两个Borker，一个为MasterBroker另一个为SlaveBroker，BrokerName相同但是BrokerId不同，SlaveBroker主要复制MasterBroker的数据）（4）、IndexService: 索引服务，根据特定的MessageKey，对投递到Broker的信息进行索引服务，同时也提供根据MessageKey对消息进行快速查询的功能 5、同步刷盘和异步刷盘 (1)、同步刷盘增加了数据的可靠性

producer -> java(通过java发送消息)  ->memory(内存) ->  disk(写入磁盘) -> memory(写入完成)->java (写入完成)

(2)、异步刷盘增加了吞吐量

producer -> java(通过java发送消息)  ->memory(内存) ->  disk(写入磁盘的同时直接返回写入成功)

6、Broker （1）、同步复制

当master和slave都成功写入数据之后才返回成功，会降低系统的吞吐量但是能够保持数据的可靠性

(2)、异步复制

当master成功写入数据之后就返回成功，降低了数据的可靠性提高了系统的吞吐量

2、生产者发送消息发送异步消息

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            Message mes = new Message("base", "tag1", ("Hello World").getBytes());
            //5、发送消息
            producer.send(mes, new SendCallback() {
                //发送成功回调函数
                @Override
                public void onSuccess(SendResult sendResult) {

                }
                //发送失败回调函数
                @Override
                public void onException(Throwable e) {

                }
            });

            //线程睡眠1秒
            TimeUnit.SECONDS.sleep(1L);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

发送同步消息

public class SyncProducer {
    public static void main(String[] args) throws  Exception{
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            Message mes = new Message("base", "tag1", ("Hello World").getBytes());
            //5、发送消息
            producer.send(mes);

            //线程睡眠1秒
            TimeUnit.SECONDS.sleep(1L);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

发送单向消息

public class OnewayProducer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            Message mes = new Message("base", "tag1", ("Hello World").getBytes());
            //5、发送消息
            producer.sendOneway(mes);

            //线程睡眠1秒
            TimeUnit.SECONDS.sleep(1L);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

发送顺序消息

public class Producer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();
        //4、模拟需要按照顺序发送消息的集合
        ArrayList list = new ArrayList();
        for (int i = 0; i < list.size(); i++) {
            String result = list.get(i).toString();
            
            Message mes = new Message("orderTopic", "orderTag", "i" + i, result.getBytes());
            
            producer.send(mes, new MessageQueueSelector() {
                
                public MessageQueue select(List mqs, Message msg, Object arg) {
                    Long orderId  = (Long) arg;
                    long index = orderId % mqs.size();
                    //选择第几个队列按照顺序发送消息
                    return mqs.get((int)index);
                }
            },list.get(i).getOrderId());
        }

    }
}

发送事物消息

public class Producer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者，并指定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        //设置事物队列的监听器
        producer.setTransactionListener(new TransactionListener() {
            
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                
                String tags = msg.getTags();
                switch (tags) {
                    case "TagA":
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case "TagB":
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    case "TagC":
                        return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //如果是丢弃的消息再在这里进行处理
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        for (int i = 0; i < 10; i++) {
            String[] arr = {"TagA", "TagB", "TagC"};

            //4、创建消息对象，指定主题Topic、Tag和消息体
            Message mes = new Message("base", arr[i], ("Hello World").getBytes());
            //5、发送消息
            producer.sendMessageInTransaction(mes, null);

            //线程睡眠1秒
            TimeUnit.SECONDS.sleep(1L);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

发送延时消息

public class Producer {
    public static void main(String[] args) throws Exception{
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            Message mes = new Message("base", "tag1", ("Hello World").getBytes());

            //设置发送延时时间
            mes.setDelayTimeLevel(5);

            //5、发送消息
            producer.send(mes);

            //线程睡眠1秒
            TimeUnit.SECONDS.sleep(1L);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

批量发送消息

public class Producer {
    public static void main(String[] args) throws Exception{
        //1、创建消息生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定NameServer,集群的NameServer用;号分隔
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、启动producer
        producer.start();

        List msgList = new ArrayList();

        Message mes1 = new Message("base", "tag1", ("Hello World").getBytes());
        Message mes2 = new Message("base", "tag2", ("Hello World").getBytes());
        Message mes3 = new Message("base", "tag3", ("Hello World").getBytes());
        msgList.add(mes1);
        msgList.add(mes2);
        msgList.add(mes3);

        SendResult send = producer.send(msgList);

        producer.shutdown();
    }
}

3、消费者普通消费

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1、创建消费者Consumer,指定消费组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2、指定NameServer地址
        consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、订阅Topic和Tag
        consumer.subscribe("base","*");

        //4、设置监听器、处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5、启动消费者Consumer
        consumer.start();
    }
}

顺序消费

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1、创建消费者Consumer,指定消费组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2、指定NameServer地址
        consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3、订阅Topic和Tag
        consumer.subscribe("orderTopic","*");

        //4、设置监听器、处理消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(msg);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
    }
}

3、整合SpringBoot

//配置文件
# nameserver
rocketmq.name-server=192.168.25.135:9576;192.168.25.138:9876
# 发送者组名
rocketmq.producer.group=my-group
# 消费者组名
rocketmq.consumer.group=my-group
//生产者
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Test
    void testSendMessage() {
        rocketMQTemplate.convertAndSend("topic","Hello SpringBoot RocketMq");
    }
//消费者
 
@RocketMQMessageListener(topic = "topic",consumerGroup = "${rocketmq.consumer.group}",consumeMode = ConsumeMode.ConCURRENTLY )
@Component
public class Consumer implements RocketMQListener {
    
    @Override
    public void onMessage(String s) {

    }
}

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5612491.html

RocketMq

发表评论

评论列表（0条）