0

0

分析Java开发RocketMQ生产者高可用示例

WBOY

WBOY

发布时间:2023-04-23 23:28:06

|

1129人浏览过

|

来源于亿速云

转载

    1 消息

    public class Message implements Serializable {
        private static final long serialVersionUID = 8445773977080406428L;
        //主题名字
        private String topic;
        //消息扩展信息,Tag,keys,延迟级别都存在这里
        private Map properties;
        //消息体,字节数组
        private byte[] body;
        //设置消息的key,
        public void setKeys(String keys) {}
        //设置topic
        public void setTopic(String topic) {}
        //延迟级别
        public int setDelayTimeLevel(int level) {}
        //消息过滤的标记
        public void setTags(String tags) {}
        //扩展信息存放在此
        public void putUserProperty(final String name, final String value) {}
    }

    消息就是孩子们,这些孩子们呢,有各自的特点,也有共性。同一个家长送来的两个孩子可以是去同一个地方的,也可以是去不同的地方的。

    1.1 topic

    首先呢,每个孩子消息都有一个属性topic,这个我们上文说到了,是一个候船大厅。孩子们进来之后,走到自己指定的候船大厅的指定区域(平时出门坐火车高铁不也是指定的站台乘车么),坐到message queue座位上等,等着出行。

    Broker有一个或者多个topic,消息会存放到topic内的message queue内,等待被消费。

    1.2 Body

    孩子消息,也有一个Body属性,这就是他的能力,他会画画,他会唱歌,他会干啥干啥,就记录在这个Body属性里。等走出去了,体现价值的地方也是这个Body属性。

    Body就是消息体,消费者会根据消息体执行对应的操作。

    立即学习Java免费学习笔记(深入)”;

    1.3 tag

    这个tag我们上节说了,就是一个标记,有的孩子背着画板,相机,有的游船就特意找到这些孩子拉走,完成他们的任务。

    可以给消息设置tag属性,消费者可以选择含有特定tag属性的消息进行消费。

    1.4 key

    key就是每个孩子消息的名字了。要找哪个孩子,喊他名就行。

    对发送的消息设置好 Key,以后可以根据这个Key 来查找消息。比如消息异常,消息丢失,进行查找会很方便。

    1.5 延迟级别

    当然,还有的孩子来就不急着走,来之前就想好了,要恰个饭,得30分钟,所以自己来了会等30分钟后被接走。

    设置延迟级别可以规定多久后消息可以被消费。

    2 生产者高可用

    每个送孩子来的家长都希望能送到候船大厅里,更不希望孩子被搞丢了,这个时候这个候船大厅就需要一些保证机制了。

    ShopNC网上商店单用户版
    ShopNC网上商店单用户版

    ShopNC单用户商城系统是面向独立卖家而开发的B2C商城系统。系统运行稳定高效,功能强大,突出个性化配置要求,可以根据不同的营销策略,从模板、栏目、功能上进行调整,满足各类客户的需要。系统部署快捷方便,减轻了使用者的技术负担,简单的维护操作免去了用户的后顾之忧。本系统前台开放源码,后台加密的。产品特点快速安装,维护简单 分布提示安装,即使不熟悉技术的用户也可以自主安装系统。后台融合数据库等功能管

    下载

    2.1 客户端保证生产者高可用

    2.1.1 重试机制

    就是说家长送来了,孩子进到候船大厅之后,没能成功坐到message queue座位上,这个时候工作人员会安排重试,再去看是否有座位坐。重试次数默认是2次,也就是说,消息孩子共有3次找座位坐的机会。

    看源码,我特意加了注解,大致可以看懂一些了。

    //这里取到了重试的次数
    int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
    int times = 0;
    String[] brokersSent = new String[timesTotal];
    for (; times < timesTotal; times++) {
        String lastBrokerName = null == mq ? null : mq.getBrokerName();
        //获取消息队列
        MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
        if (mqSelected != null) {
            mq = mqSelected;
            brokersSent[times] = mq.getBrokerName();
            try {
                beginTimestampPrev = System.currentTimeMillis();
                if (times > 0) {
                    //Reset topic with namespace during resend.
                    msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));
                }
                long costTime = beginTimestampPrev - beginTimestampFirst;
                if (timeout < costTime) {
                    callTimeout = true;
                    break;
                }
                //发送消息
                sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
                ...
            } catch (RemotingException e) {
                ...
                continue;
            } catch (MQClientException e) {
                ...
                continue;
            } catch (MQBrokerException e) {
                ...
                continue;
            } catch (InterruptedException e) {
                //可以看到只有InterruptedException抛出了异常,其他的exception都会继续重试
                throw e;
            }
        } else {
            break;
        }
    }

    重试代码如上,这个sendDefaultImpl方法中,会尝试发送三次消息,若是都失败,才会抛出对应的错误。

    2.1.2 客户端容错

    若是有多个Broker候车大厅的时候,服务人员会安排消息孩子选择一个相对不拥挤比较容易进入的来进入。当然那些已经关闭的停电的没有服务能力的,我们是不会进的。

    MQ Client会维护一个Broker的发送延迟信息,根据这个信息会选择一个相对延迟较低的Broker来发送消息。会主动剔除哪些已经宕机,不可用或发送延迟级别较高的Broker.

    选择Broker就是在选择message queue,对应的代码如下:

    这里会先判断延迟容错开关是否开启,这个开关默认是关闭的,若是开启的话,会优先选择延迟较低的Broker。

    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        //判断发送延迟容错开关是否开启
        if (this.sendLatencyFaultEnable) {
            try {
                //选择一个延迟上可以接受,并且和上次发送相同的Broker
                int index = tpInfo.getSendWhichQueue().incrementAndGet();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    //若是Broker的延迟时间可以接受,则返回这个Broker
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
                        return mq;
                }
                //若是第一步没能选中一个Broker,就选择一个延迟较低的Broker
                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
                    }
                    return mq;
                } else {
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }
            //若是前边都没选中一个Broker,就随机选一个Broker
            return tpInfo.selectOneMessageQueue();
        }
        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }

    但是当延迟容错开关关闭状态的时候,执行的代码如下:

    为了均匀分散Broker的压力,会选择与之前不同的Broker

    public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
        //若是没有上次的Brokername做参考,就随机选一个
        if (lastBrokerName == null) {
            return selectOneMessageQueue();
        } else {
            //如果有,那么就选一个其他的Broker
            for (int i = 0; i < this.messageQueueList.size(); i++) {
                int index = this.sendWhichQueue.incrementAndGet();
                int pos = Math.abs(index) % this.messageQueueList.size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = this.messageQueueList.get(pos);
                //这里判断遇上一个使用的Broker不是同一个
                if (!mq.getBrokerName().equals(lastBrokerName)) {
                    return mq;
                }
            }
            //若是上边的都没选中,那么就随机选一个
            return selectOneMessageQueue();
        }
    }

    2.2 Broker端保证生产者高可用

    Broker候船大厅为了能确切的接收到消息孩子,至少会有两个厅,一个主厅一个副厅,一般来说孩子都会进入到主厅,然后一顿操作,卡该忙信那机资(影分身之术),然后让分身进入到副厅,这样当主厅停电了,不工作了,副厅的分身只要去完成了任务就ok的。一般来说都是主厅的消息孩子去坐船完成任务。

    相关文章

    java速学教程(入门到精通)
    java速学教程(入门到精通)

    java怎么学习?java怎么入门?java在哪学?java怎么学才快?不用担心,这里为大家提供了java速学教程(入门到精通),有需要的小伙伴保存下载就能学习啦!

    下载

    相关标签:

    本站声明:本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

    相关专题

    更多
    excel制作动态图表教程
    excel制作动态图表教程

    本专题整合了excel制作动态图表相关教程,阅读专题下面的文章了解更多详细教程。

    20

    2025.12.29

    freeok看剧入口合集
    freeok看剧入口合集

    本专题整合了freeok看剧入口网址,阅读下面的文章了解更多网址。

    65

    2025.12.29

    俄罗斯搜索引擎Yandex最新官方入口网址
    俄罗斯搜索引擎Yandex最新官方入口网址

    Yandex官方入口网址是https://yandex.com;用户可通过网页端直连或移动端浏览器直接访问,无需登录即可使用搜索、图片、新闻、地图等全部基础功能,并支持多语种检索与静态资源精准筛选。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    197

    2025.12.29

    python中def的用法大全
    python中def的用法大全

    def关键字用于在Python中定义函数。其基本语法包括函数名、参数列表、文档字符串和返回值。使用def可以定义无参数、单参数、多参数、默认参数和可变参数的函数。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    16

    2025.12.29

    python改成中文版教程大全
    python改成中文版教程大全

    Python界面可通过以下方法改为中文版:修改系统语言环境:更改系统语言为“中文(简体)”。使用 IDE 修改:在 PyCharm 等 IDE 中更改语言设置为“中文”。使用 IDLE 修改:在 IDLE 中修改语言为“Chinese”。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    16

    2025.12.29

    C++的Top K问题怎么解决
    C++的Top K问题怎么解决

    TopK问题可通过优先队列、partial_sort和nth_element解决:优先队列维护大小为K的堆,适合流式数据;partial_sort对前K个元素排序,适用于需有序结果且K较小的场景;nth_element基于快速选择,平均时间复杂度O(n),效率最高但不保证前K内部有序。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    12

    2025.12.29

    php8.4实现接口限流的教程
    php8.4实现接口限流的教程

    PHP8.4本身不内置限流功能,需借助Redis(令牌桶)或Swoole(漏桶)实现;文件锁因I/O瓶颈、无跨机共享、秒级精度等缺陷不适用高并发场景。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    134

    2025.12.29

    抖音网页版入口在哪(最新版)
    抖音网页版入口在哪(最新版)

    抖音网页版可通过官网https://www.douyin.com进入,打开浏览器输入网址后,可选择扫码或账号登录,登录后同步移动端数据,未登录仅可浏览部分推荐内容。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    63

    2025.12.29

    快手直播回放在哪看教程
    快手直播回放在哪看教程

    快手直播回放需主播开启功能才可观看,主要通过三种路径查看:一是从“我”主页进入“关注”标签再进主播主页的“直播”分类;二是通过“历史记录”中的“直播”标签页找回;三是进入“个人信息查阅与下载”里的“直播回放”选项。本专题为大家提供相关的文章、下载、课程内容,供大家免费下载体验。

    18

    2025.12.29

    热门下载

    更多
    网站特效
    /
    网站源码
    /
    网站素材
    /
    前端模板

    精品课程

    更多
    相关推荐
    /
    热门推荐
    /
    最新课程
    Kotlin 教程
    Kotlin 教程

    共23课时 | 2.1万人学习

    C# 教程
    C# 教程

    共94课时 | 5.6万人学习

    Java 教程
    Java 教程

    共578课时 | 39.6万人学习

    关于我们 免责申明 举报中心 意见反馈 讲师合作 广告合作 最新更新
    php中文网:公益在线php培训,帮助PHP学习者快速成长!
    关注服务号 技术交流群
    PHP中文网订阅号
    每天精选资源文章推送

    Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号