<body> manager.cs using System; using System.Collections.Concurrent; using System.Threading; using bntu.pcm.plworker; using bntu.pcm.works; /* * bntu 是我的网名，在cnblogs上可以搜到我的博客：https://www.cnblogs.com/SimbaWang/ * pcm 是producer/consumer man
消息队列 - 基础篇 目录消息队列 - 基础篇前言消息模型消息丢失消息丢失检测消息可靠传递消息重复服务质量标准用幂等性解决消息重复消息积压Producer 性能Consumer 性能消息积压排查分布式事物事务消息参阅 前言 常见问题： 技术选型 高可靠、高可用、高性能 不重复、不丢失 水平
如何保证消息的顺序性？ 面试官心理分析 其实这个也是用 MQ 的时候必问的话题，第一看看你了不了解顺序这个事儿？第二看看你有没有办法保证消息是有顺序的？这是生产系统中常见的问题。 面试题剖析 我举个例子，我们以前做过一个 mysql binlog 同步的系统，压力还是非常大的，日同步数
如何解决消息队列的延时以及过期失效问题？消息队列满了以后该怎么处理？有几百万消息持续积压几小时，说说怎么解决？ 面试官心理分析 你看这问法，其实本质针对的场景，都是说，可能你的消费端出了问题，不消费了；或者消费的速度极其慢。接着就坑爹了，可能你的消息队列集群的磁盘都快写满了，
Kafka有哪几个部分组成 生产者、消费者、topic、group、partition kafka的group1）定义：即消费者组是 Kafka 提供的可扩展且具有容错性的消费者机制。在Kafka中，消费者组是一个由多个消费者实例构成的组。同一个组下的每个实例都配置有相同的组ID，被分配不同的订阅分区。当某个实例挂掉
常见方法 流模型的操作很丰富，这里介绍一些常用的API，这些方法可以被分成两种： 延迟方法:返回值类型仍然是Stream接口自身类型的方法 因此支持链式调用（除了终结方法外 其余方法均为延迟方法。） 终结方法:返回值类型不再是Stream接口自身类型的方法 因此不再支持类似StringBu
Kafka生产者与消费者 1. kafka客户端——生产者 1. pom配置 <properties> <lombok.version>1.16.18</lombok.version> <fastjson.version>1.2.66</fastjson.version> <kafka.version>2.4.1</kafka.version> </pr
背景说明 HSF是闭源的，考虑用开源产品（dubbo）进行替代。 如果是考虑要从一个rpc框架转成另一个，或许也可以参考本文。 主要思想：进行rpc的发布、订阅操作，其实是集中在2个类里面（provider/consumer），而不是散落在每个实现类里面。而替换成其它rpc时候，就是针对父类（providerFather/consumerF
assume [from ad- 'to' + sumere 'to take'] take it for granted that: believe that sh is true without making sure consume [from consumere, from com- + sumere 'to take up, take'] resume [from resumere, from sumere 'to ta
【问题背景】 快应用A中下载原生应用A是不允许的，但涉及部分场景，需要安装原生APP。 【解决方法】 快应用提供pkg.install(OBJECT)接口可以跳转到应用市场的安装详情页面，引导用户安装原生应用。 接口链接： https://developer.huawei.com/consumer/cn/doc/development/quickApp-Ref
今天我们来分析一下这个问题。 先来回忆一下kafka 中消息传输的整个过程 1、kafka 在producer 端产生消息，调用kafka producer client send方法发送消息 2、kafka producer client 使用一个单独的线程，异步的将消息发送给kafka server 3、kafka server收到消息以后，保存数据，并同
Supplier接口 Supplier在英文中的意思：供应商。 指提供某种产品。编程中一般认为是生成某种类型的数据。 jdk源码 package java.util.function; @FunctionalInterface public interface Supplier<T> { T get(); } 除了一个抽象方法以外，并没有其它的方法 我们来看看这个接口的
Kafka消息交付可靠性保障 Kafka 消息交付可靠性保障以及精确处理一次语义的实现。 Kafka 对 Producer 和 Consumer 提供的消息交付可靠性保障： 最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。 至少一次（at least once）：消息不会丢失，但有可能被重复发送。 精确一次（exactly o
Kafka 消费者组 Consumer Group 是kafka 当中比较有亮点的设计了。 1. 消费者组的特点： 一个分区只可以被消费组中的一个消费者所消费。 一个消费组中的一个消费者可以消费多个分区，例如 C1 消费了 P0, P3。 一个消费组中的不同消费者消费的分区一定不会重复，例如： C1 -> P0、P3 C2
kafka 位移主题 __consumer_offsets 位移主题 诞生背景 老版本的Kafka会把位移信息保存在Zookeeper中，当Consumer重启后，自动从Zk中读取位移信息。这种设计使Kafka Broker不需要保存位移数据，可减少Broker端需要持有的状态空间，有利于实现高伸缩性。 Zookeeper不适用于高频的写操作，这
kafka 提交位移 Kafka Consumer 提交位移的方式有两种：自动提交位移和手动提交位移(包括同步提交与异步提交)。 位移提交的语义保障是由你来负责的，Kafka 只会“无脑”地接受你提交的位移。你对位移提交的管理直接影响了你的 Consumer 所能提供的消息语义保障。 自动提交位移 ena
一、offset 管理 这里的 offset 指的是 Consumer 的消费进度 offset。 消费进度 offset 是用来记录每个 Queue 的不同消费组的消费进度的。根据消费进度记录器的不同，可以分为两种模式：本地模式和远程模式。 二、offset 本地管理模式 当消费模式为广播消费时，offset 使用本地模式存
消息中间件 常用消息中间件对比 RocketMQ相比其他两种有一个不可替代的优势：支持事务消息 作用 异步解耦（提高效率） 削峰填谷（缓解压力） 分布式缓存同步 消息分发 写磁盘方式 RocketMQ的消息是存储到磁盘上的，这样既能保证断电后恢复，又可以让存储的消息量超出内存的限制。Rocke
1、kafka数据分区和消费者的关系：1个partition只能被同组的⼀个consumer消费，同组的consumer则起到均衡效果 2、kafka的数据offset读取流程 1.连接ZK集群，从ZK中拿到对应topic的partition信息和partition的Leader的相关信息 2.连接到对应Leader对应的broker 3.consume
Java高级特性——平台模块系统 java9 引入了 JAVA 平台模块系统 一、模块的概念 假设一个场景，我们想要修改或剔除一个很少使用的特性，如果它对外是公有的，那么就没办法推断这个变化产生的影响。模块为我们提供了比较好的解决方案。 一个 JAVA 平台模块包含： 一个包集合 可选地包含资
1. 依赖 <!-- kafka client --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> 2. 配置文件配置kafka bootstrap-server ######## kafka client #########
1、KafKa是什么？ Kafka 是一种高吞吐量、分布式、基于发布/订阅的消息系统，最初由 LinkedIn 公司开发，使用Scala 语言编写，目前是 Apache 的开源项目。1. broker： Kafka 服务器，负责消息存储和转发2. topic：消息类别， Kafka 按照 topic 来分类消息3. partition： topic 的
1. kafka-python的安装 pip3 install kafka-python 2.kafka-python的基本使用 最简单使用实例 1.消费端 from kafka import KafkaConsumer consumer = KafkaConsumer('my_topic', group_id= 'group2', bootstrap_servers= ['localhost:9092']，value_seriali
先看顺序会错乱的场景：RabbitMQ，一个queue，多个consumer，这不明显乱了； 解决： 1、拆分多个queue，每个queue一个consumer，就是多一些queue而已，确实有点麻烦， 2、或者就一个queue就是对应一个consumer，然后这个consumer内部调用内存队列做排队，然后分发给底层不同的worker来处理。
消息积压处理办法：临时紧急扩容：先修复 consumer 的问题，确保其恢复消费速度，然后将现有 cnosumer 都停掉。新建一个 topic，partition 是原来的 10 倍，临时建立好原先 10 倍的 queue 数量。然后写一个临时的分发数据的 consumer 程序，这个程序部署上去消费积压的数据，消费之
