剖析producer之前，我们来回顾一下Kafka的producer，producer（生产者）：消息放到队列里面的叫生产者。
“Pulsar is a distributed pub-sub messaging platform with a very flexible messaging model and an intuitive client API.”
上篇文章说了，kafka需要先构造properties指定server和kafka集群，key 和 value用stringSerialize序列化，通过producer发送send，需要records参数指定topic和value，之后发送消息，有异步和同步，最后关闭。
本文主要研究一下rocketmq-client-go的transactionProducer
生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信，生产者生产消费者需要的资料，消费者把资料做成产品。
queue模块的Queue对象实现了多生产者/多消费者队列，尤其适合需要在多个线程之间进行信息交换的场合，实现了多线程编程所需要的所有锁语义。 Queue对象主要实现了put()和get()方法，分别用来往队列尾部追加元素和在队列头部获取并删除元素。这两个方法都允许指定超时时间，其用法分别为put(item, block=True, timeout=None)和get(block=True, timeout=None) 在下面的代码中，自定义了生产者线程类和消费者线程类，生产者生产随机数量个元素并放置到队列
RocketMQ可用于以三种方式发送消息：同步、异步和单向传输。前两种消息类型是可靠的，因为无论它们是否成功发送都有响应。
Kafka只对“已提交”的消息（committed message）做有限度的持久化保证。
如果Kafka Producer使用“发后即忘”的方式发送消息，即调用producer.send(msg)方法来发送消息，方法会立即返回，但此时并不能说明消息已经发送成功。消息发送方式详见初次邂逅Kafka生产者。
这几天很忙，但是我现在给我的要求是一周至少要出一篇文章，所以先拿这篇笔记来做开胃菜，源码分析估计明后两天应该能写一篇。给自己加油~，即使没什么人看。
修改kafka的配置文件server.properties,按照自己的IP和主机名称修改下面的配置并且打开
kafka0.9版本以后用java重新编写了producer，废除了原来scala编写的版本。
2. yield()方法的意思是告诉CPU执行其他的线程，当前线程让出CPU的执 行权利
通过三种方式来发送RocketMQ消息使用: 可靠的同步发送, 可靠的异步发送和单向传输。
上篇文章说了，kafka新版旧版的区别，producer全部异步发消息，并且提供回调机制callback，判断是否成功，通过分批次发送batching保证吞吐量，分区策略更加合理，旧版本默认是在一段时间内把消息发到固定区域，新版本采用轮询，消息更加均匀。Consumer新版为单线程执行，单个consumer线程管理多个socker，在10版本后，加入了心跳线程，这最多也就算了是双线程。偏移量 在新版本交给kafka处理，舍弃了zookeeper，这样可以依赖kafka备份机制天然实现高可用原理。
本章总结 : 使用了 泛型 out 协变 和 泛型 in 逆变 极大的提高了程序的扩展性 ;
最近和一些同学交流的时候反馈说，在面试Kafka时，被问到Kafka组件组成部分、API使用、Consumer和Producer原理及作用等问题都能详细作答。但是，问到一个平时不注意的问题，就是Kafka的幂等性，被卡主了。那么，今天笔者就为大家来剖析一下Kafka的幂等性原理及实现。
Kafka旧版本producer由scala编写，0.9以后已经废除，但是很多公司还在使用0.9以前的版本，所以总结如下： 要注意包Producer是 kafka.javaapi.producer.Producer 这个才是java api使用的包
GET /ecommerce/product/_search { "query": { "match": { "name": "yagao" //查询包含单词 } }, "_source": ["name","price"],//不写查询所有字段 "sort": [ { "price": { "order": "desc"//倒序排序 } } ], "from": 0,//分页 "size": 1 }
上面3种EOS语义有着不同的应用范围，幂等producr只能保证单分区上无重复消息；事务可以保证多分区写入消息的完整性；而流处理EOS保证的是端到端(E2E)消息处理的EOS。用户在使用过程中需要根据自己的需求选择不同的EOS。以下是启用方法：
本文主要解析一下spring for apache kafka对原生的kafka client producer的封装与集成。
写在前面的话 本文所有Kafka原理性的描述除特殊说明外均基于Kafka 1.0.0版本。 为什么要提供事务机制 Kafka事务机制的实现主要是为了支持 Exactly Once即正好一次语义 操作的原子性 有状态操作的可恢复性 Exactly Once 《Kafka背景及架构介绍》一文中有说明Kafka在0.11.0.0之前的版本中只支持At Least Once和At Most Once语义，尚不支持Exactly Once语义。 但是在很多要求严格的场景下，如使用Kafka处理交易数据，Exactl
遇到的坑，一开始报的错误莫名其妙，一开始以为使用的jar包版本问题，又是报slf4j的错误，又是报log4j的错误，又是报空指针的异常。最后百度意外遇到了可能是本地没有将ip地址放到hosts文件里面，果然是这个问题。
# 9.py #code=utf-8 # python的协程使用 ''' 所以子程序调用是通过栈实现的，一个线程就是执行一个子程序。 Python对协程的支持还非常有限，用在generator中的yield可以一定程度上实现协程。虽然支持不完全，但已经可以发挥相当大的威力了。 Python通过yield提供了对协程的基本支持，但是不完全。而第三方的gevent为Python提供了比较完善的协程支持。 由于gevent是基于IO切换的协程，所以最神奇的是，我们编写的Web App代码，不需要引入gevent的包，也不需要改任何代码，仅仅在部署的时候，用一个支持gevent的WSGI服务器，立刻就获得了数倍的性能提升。 ''' import time def consumer(): r = '' while True: n = yield r if not n: return print('[CONSUMER] Consuming %s...' % n) time.sleep(1) r = '200 ok' def produce(c): c.next() n = 0 while n < 5: n = n + 1 print('[PRODUCER] Producing %s...' % n) r = c.send(n) print('[PRODUCER] Consumer return: %s' % r) c.close() c = consumer() produce(c) ''' 上面程序逻辑是： 首先调用c.next()启动生成器； 然后，一旦生产了东西，通过c.send(n)切换到consumer执行； consumer通过yield拿到消息，处理，又通过yield把结果传回； produce拿到consumer处理的结果，继续生产下一条消息； produce决定不生产了，通过c.close()关闭consumer，整个过程结束。 ''' ''' 执行结果是 [PRODUCER] Producing 1... [CONSUMER] Consuming 1... [PRODUCER] Consumer return: 200 ok [PRODUCER] Producing 2... [CONSUMER] Consuming 2... [PRODUCER] Consumer return: 200 ok [PRODUCER] Producing 3... [CONSUMER] Consuming 3... [PRODUCER] Consumer return: 200 ok [PRODUCER] Producing 4... [CONSUMER] Consuming 4... [PRODUCER] Consumer return: 200 ok [PRODUCER] Producing 5... [CONSUMER] Consuming 5... [PRODUCER] Consumer return: 200 ok '''
之前和大家聊过kafka是如何保证消息不丢失的，今天再讲讲在不丢消息的同时，如何实现精确一次处理的语义实现。
# apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞
在前面文章的producer和consumer示例中，只存在一个进程。producer调用put时执行了consumer中的put方法，consumer调用get时执行了producer中的get方法。
Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。
Producer拦截器(interceptor)是在Kafka 0.10版本被引入的，主要用于实现clients端的定制化控制逻辑。 对于producer而言，interceptor使得用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：
Java生产者消费者是最基础的线程同步问题，java岗面试中还是很容易遇到的，之前没写过多线程的代码，面试中被问到很尬啊，面完回来恶补下。在网上查到大概有5种生产者消费者的写法，分别如下。
这篇文章主要讲述 Kafka 事务性的实现，这部分的实现要比幂等性的实现复杂一些，幂等性实现是事务性实现的基础，幂等性提供了单会话单 Partition Exactly-Once 语义的实现，正是因为 Idempotent Producer 不提供跨多个 Partition 和跨会话场景下的保证，因此，我们是需要一种更强的事务保证，能够原子处理多个 Partition 的写入操作，数据要么全部写入成功，要么全部失败，不期望出现中间状态。这就是 Kafka Transactions 希望解决的问题，简单来说就是能够实现 atomic writes across partitions，本文以 Apache Kafka 2.0.0 代码实现为例，深入分析一下 Kafka 是如何实现这一机制的。
这篇文章主要讲述 Kafka 事务性相关原理，从 Kafka EOS 语义、幂等性、事务性等几个方面阐述。
http://rocketmq.apache.org/dowloading/releases/
我写了一点逗比代码，让在每次使用 NewLife 的 RocketMQ 发送消息时，都创建一个新的 Producer 生产者。此时我发现了在我的消费者里面，无论开多少个消费者实例进程，每次都只有一个消费者进行消费
在平时对kafka的运维工作中，我们经常会由于某些原因去删除一个topic，比如这个topic是测试用的，生产环境中需要删除。或者我想扩容topic的同时，这个topic中的数据我不想要了，这时候删除topic，增加broker，再重新创建topic就会是比较简单的方法。但是kafka删除topic时，有很多关键的点必须清楚，否则在删除topic的时候就会出现各种各样的问题。
为了解决重试导致的消息重复、乱序问题，kafka引入了幂等消息。幂等消息保证producer在一次会话内写入一个partition内的消息具有幂等性，可以通过重试来确保消息发布的Exactly Once语义。
在上文中介绍了AdminClient API的使用，现在我们已经知道如何在应用中通过API去管理Kafka了。但在大多应用开发中，我们最常面临的场景就是发送消息到Kafka，或者从Kafka中消费消息，也就是典型的生产/消费模式。而本文将要演示的就是如何使用Producer API将消息发送至Kafka中，使应用成为一个生产者。
TransferQueue(java7引入)继承了BlockingQueue（BlockingQueue又继承了Queue）并扩展了一些新方法。生产者会一直阻塞直到所添加到队列的元素被某一个消费者所消费（不仅仅是添加到队列里就完事）。
源节点就是第一个EOS节点（Genesis node），也可以叫主节点，EOS多节点组网的前提是已经对单机环境非常熟悉，我们的架构如下：
原文链接：https://www.cnblogs.com/Evsward/p/eos-boot.html
Kafka 事务与数据库的事务定义基本类似，主要是一个原子性：多个操作要么全部成功，要么全部失败。Kafka 中的事务可以使应用程序将消费消息、生产消息、提交消费位移当作原子操作来处理。 为了实现事务，Producer 应用程序必须做到：
举例：卖东西 前提：有一个接口IProducer，一个实现类Producer（最后我会附上这部分的代码（“动态代理介绍”） IProducer proxyProducer为动态代理对象，方法执行调用它
maxwell-1.25.1/src/main/java/com/zendesk/maxwell/bootstrap/BootstrapController.java
对于上面的例子，天气预报需要用户自己付费订阅、群聊需要先进群、在家看报需要联系报社订阅报纸。这里就能看出来。这种关系是一个一对多的关系。被观察者是同一个，而观察者却可以是很多个不同的对象。还有就是观察者需要自己主动的去找被观察者“提前”说明好，“一旦有消息，请通知我一声”。所有这里可以抽象出来几个角色和动作。
一般在KafKa消费程序中消费可以设置多个主题，那在同一程序中需要向KafKa发送不同主题的消息，如异常需要发到异常主题，正常的发送到正常的主题，这时候就需要实例化多个主题，然后逐个发送。 在
根据RabbitMQ的工作模式，一条消息从生产者发出，到消费者消费，需要经历以下4个步骤：
wait和notify以及notifyAll之所以是Object的方法就是因为任何一个对象都可以当做锁对象（锁对象也是一种临界资源）
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
