远程连接 kafka 配置

默认的 kafka 配置是无法远程访问的，解决该问题有几个方案。

方案1

advertised.listeners=PLAINTEXT://IP:9092

注意必须是 ip，不能是 hostname

方案2

advertised.listeners=PLAINTEXT://node0:9092

node0 是 hostname，需在 /etc/hosts 中 添加一行

172.16.89.80 node0

然后 必须在 远程机（要访问 kafka 的机器 windows）上修改 hosts文件，

C:\Windows\System32\drivers\etc\hosts

在末尾加上

IP1  节点1

IP2  节点2

节点名与服务器上的 hostname 相同。

测试异常记录

WARN [Consumer clientId=consumer-1, groupId=console-consumer-4184] Connection to node -1 (/172.16.89.80:9092) could not be established. Broker may not be available

. (org.apache.kafka.clients.NetworkClient)

kafka 配置 与 console 启动的 ip 不一致，如 配置文件中 listeners=PLAINTEXT://172.16.89.80:9092，启动 是  localhost

kafka.errors.NoBrokersAvailable: NoBrokersAvailable

listeners=PLAINTEXT://172.16.89.80:9092

基础操作

最简单的场景，生产者发送，消费者接收

Producer

send(self, topic, value=None, key=None, headers=None, partition=None, timestamp_ms=None)

producer = KafkaProducer(bootstrap_servers = '172.16.89.80:9092')

# producer = KafkaProducer(bootstrap_servers = 'node0:9092')

# 如果这里是 ip，后面必须加 producer.flush() 或者 producer.close()；

# 如果是 hostname，则不需要，但是好像有丢包

print(producer.config)  # 打印配置信息

topic = ''

for i in range(200):

    msg = "msg%d" % i

    producer.send(topic, msg)

producer.flush()

# producer.close()

上面的注释是我亲测的结果，至于为什么，我还没明白，谁能帮我解答

Consumer

topic = ''

consumer = KafkaConsumer(topic, bootstrap_servers=['172.16.89.80:9092'])

# consumer是一个消息队列，当后台有消息时，这个消息队列就会自动增加．所以遍历也总是会有数据，当消息队列中没有数据时，就会堵塞等待消息带来

for msg in consumer:

    recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)

    print recv

    time.sleep(2)

注意 这种消费方式只能消费实时数据，不能消费历史数据

如果想读历史消息，可以这样写

consumer = KafkaConsumer(topic, auto_offset_reset='earliest', bootstrap_servers=['172.16.89.80:9092'])

auto_offset_reset：重置 offset，earliest 表示移到最早的可用消息，lastest 为最新消息，默认 latest

源码定义:{‘smallest’: ‘earliest’, ‘largest’: ‘latest’}

输出

生产者不指定分区，消费者输出如下

91101:1:8: key=None value=msg0

91101:2:12: key=None value=msg1

91101:0:17: key=None value=msg2

3 条消息分到了3个分区

如果不指定分区，一个 topic 的多个消息会被分发到不同的分区；消费者也收到了所有分区的消息

生产者指定分区，消费者输出如下

91101:2:13: key=None value=msg0

91101:2:14: key=None value=msg1

91101:2:15: key=None value=msg2

3 条消息被分到同一个分区

阻塞发送

kafka send 消息是异步的，即使 send 发生错误，程序也不会提示，我们可以通过 阻塞 的方式确认是否发送。

method1-get

import pickle

import time

from kafka import KafkaProducer

from kafka.errors import kafka_errors

producer = KafkaProducer(

    bootstrap_servers=['172.16.89.80:9092'],

    key_serializer=lambda k: pickle.dumps(k),

    value_serializer=lambda v: pickle.dumps(v))

start_time = time.time()

for i in range(0, 100):

    future = producer.send(topic="", key="num", value=i)

    # 同步阻塞,通过调用get()方法保证有序性.

    try:

        record_metadata = future.get(timeout=20)

        # print(record_metadata.topic)

        # print(record_metadata.partition)

        # print(record_metadata.offset)

    except kafka_errors as e:

        print(str(e))

end_time = time.time()

time_counts = end_time - start_time

print(time_counts)

method2-flush

import pickle

import time

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['172.16.89.80:9092'],

                         key_serializer=lambda k: pickle.dumps(k),

                         value_serializer=lambda v: pickle.dumps(v))

start_time = time.time()

for i in range(0, 100):

    future = producer.send('', key='num', value=i, partition=0)

# 将缓冲区的全部消息push到broker当中

producer.flush()

producer.close()

end_time = time.time()

time_counts = end_time - start_time

print(time_counts)


格式化输入输出


KafkaProducer 指定生产者输入的格式转换方式，key_serializer、value_serializer 用于格式化 key 和 value


KafkaConsumer 指定消费者输出的格式转换方式，value_deserializer


指定消费者组


之前讲过，不同消费者组消费相同的 topic，互不影响；同一个消费者组的不同成员在同一时刻不能消费同一个 topic 相同的分区；一个线程相当于一个消费者　　　　　　【验证】


这里来验证一下


Producer


不指定分区，20条消息，3个分区




producer = KafkaProducer(bootstrap_servers = '172.16.89.80:9092')

print(producer.config)##打印配置信息

topic = ''

for i in range(20):

    msg = "msg%d" % i

    producer.send(topic, msg)

    time.sleep(2)

producer.close()




Consumer


开两个线程，或者两个窗口，指定相同的消费者组，数组名 随便写，一样即可；


也是实时消费；




topic = ''

consumer = KafkaConsumer(topic, group_id='', bootstrap_servers=['172.16.89.80:9092'])

for msg in consumer:

    recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)

    print recv




输出


一个消费者取到 0和1 分区的 value




91101:1:28: key=None value=msg1

91101:0:51: key=None value=msg3

91101:0:52: key=None value=msg5

91101:0:53: key=None value=msg9

91101:0:54: key=None value=msg10

91101:1:29: key=None value=msg11

91101:0:55: key=None value=msg12

91101:1:30: key=None value=msg15

91101:0:56: key=None value=msg16

91101:1:31: key=None value=msg17

91101:0:57: key=None value=msg18




一个消费者取到 2 分区的 value




91101:2:46: key=None value=msg0

91101:2:47: key=None value=msg2

91101:2:48: key=None value=msg4

91101:2:49: key=None value=msg6

91101:2:50: key=None value=msg7

91101:2:51: key=None value=msg8

91101:2:52: key=None value=msg13

91101:2:53: key=None value=msg14

91101:2:54: key=None value=msg19




也就是没有消费不同的分区，结论正确。


由于只有2个消费者，3个分区，所以必须有一个消费者消费2个分区；


如果是3个消费者应该就不存在这种情况，经验证，确实如此；


如果有再多的消费者，就分不到消息了；


这也验证了 之前讲的，同一个topic，不推荐多于 partition 个数的 消费者来消费，会造成资源浪费。　　　　　　　　【验证】


小结


1. 消费者不指定组，能收到所有分区的 消息


2. 如果指定了组，同组的不同消费者会消费不同的分区


3. 如果2个分区2个消费者，则一人一个分区 ；如果2个分区3个消费者，则有一个人收不到消息；


4. 如果想消费同一分区，指定不同的组


这种特性可以用作 负载均衡


设定 offset


kafka 提供了 偏移量 offset 的概念，根据偏移量可以读取 终端开启前 未接收的数据， 也可以读取任意位置的数据　　【可读取历史数据】


为了验证效果，操作如下


1. 创建一个新的 topic


2. 开启生产者，不开启消费者，并发送一些数据


3. 开启消费者，并指定 分区 和 偏移量，设置偏移量为 0


4. 再次运行消费者，指定相同的分区，偏移量设置为 4




topic = ''

consumer = KafkaConsumer(group_id='', bootstrap_servers=['172.16.89.80:9092'])

# consumer 指定主题和分区

consumer.assign([TopicPartition(topic, partition=0), TopicPartition(topic, partition=1)])       #

# 获取主题的分区信息，

print consumer.partitions_for_topic(topic)  # None

# 获取 consumer 的指定

print consumer.assignment() # set([TopicPartition(topic='91107', partition=0), TopicPartition(topic='91107', partition=1)])

# 获取 consumer 指定分区 的 起始offset

print consumer.beginning_offsets(consumer.assignment())

# {TopicPartition(topic=u'91107', partition=0): 0, TopicPartition(topic=u'91107', partition=1): 0}

consumer.seek(TopicPartition(topic, partition=1), 4)

for msg in consumer:

    recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)

    print recv




输出




# offset 0

91107:1:0: key=None value=msg2

91107:1:1: key=None value=msg6

91107:1:2: key=None value=msg10

91107:1:3: key=None value=msg11

91107:1:4: key=None value=msg13

91107:1:5: key=None value=msg19

# offset 4

91107:1:4: key=None value=msg13

91107:1:5: key=None value=msg19




相当于一个文件中从头到尾 6 行，offset 4 从第 4 行开始读，【行数从 0 开始】


异常记录




kafka.errors.IllegalStateError: IllegalStateError: You must choose only one way to configure your consumer: (1) subscribe to specific topics by name, (2) subscribe to topics matching a regex pattern, (3) assign itself specific topic-partitions.




KafkaConsumer 和 assign 不能同时 指定 topic


小结


1. offset 相当于文件中的 offset 概念


2. 指定 offset 时，必须指定分区，一个分区相当于一个文件，指定分区就相当于指定文件，offset 表示 从文件中 offset 行开始读


3. offset 功能可以读取 历史数据


定时主动拉取


主动拉取可能拉不到




from kafka import KafkaConsumer

import time

# consumer = KafkaConsumer(bootstrap_servers=['node0:9092'])      # 这样写只能获取最新消息

consumer = KafkaConsumer(bootstrap_servers=['node0:9092'], auto_offset_reset='earliest')        # 这样可以从头拉取

consumer.subscribe(topics=('',''))

while True:

    msg = consumer.poll(timeout_ms=5)   # 从kafka获取消息

    print(msg.keys())

    print('*' * 100)

    time.sleep(2)




主动从多个 topic 拉取数据


输出




[]

****************************************************************************************************

[TopicPartition(topic=u'', partition=1), TopicPartition(topic=u'', partition=0)]

****************************************************************************************************

[TopicPartition(topic=u'', partition=0), TopicPartition(topic=u'', partition=2)]

****************************************************************************************************

[TopicPartition(topic=u'', partition=1), TopicPartition(topic=u'', partition=2)]




可以看到有的回合没有拉倒数据


消息挂起与恢复


挂起，消费者不能消费，恢复后，才能消费




from kafka import KafkaConsumer

from kafka.structs import TopicPartition

import time

topic = ''

consumer = KafkaConsumer(bootstrap_servers=['node0:9092'])

consumer.subscribe(topics=(topic))

consumer.topics()

consumer.pause(TopicPartition(topic=topic, partition=0))  # pause执行后，consumer不能读取，直到调用resume后恢复。

num = 0

while True:

    print(num)

    print(consumer.paused())   # 获取当前挂起的消费者

    msg = consumer.poll(timeout_ms=5)

    print(msg)

    time.sleep(2)

    num = num + 1

    if num == 10:

        print("resume...")

        consumer.resume(TopicPartition(topic=topic, partition=0))

        print("resume......")




参考资料：


https://www.cnblogs.com/tigerzhouv587/p/11232398.html　　python 发送kafka


https://kafka-python.readthedocs.io/en/master/usage.html


https://blog.csdn.net/luanpeng825485697/article/details/81036028  　　好全的资料


https://www.jianshu.com/p/776c188cefa9　　从 zookeeper 消费


https://cloud.tencent.com/developer/news/202116　　也挺全的资料
												


																

最新文章
	

    
						
  1. tomee 消息持久化
		
    2. 
						
  2. css 层的嵌套
		
    3. 
						
  3. linux基础命令学习（六）DHCP服务器配置
		
    4. 
						
  4. Spring3.1新属性管理API：PropertySource、Environment、Profile
		
    5. 
						
  5. C#学习笔记（七）：智能编译器
		
    6. 
						
  6. MyBatis(3.2.3) - One-to-one mapping using nested Select
		
    7. 
						
  7. jquery ajax 访问webServer的xml文件
		
    8. 
						
  8. debian 显示器使用自定义分辨率
		
    9. 
						
  9. Java Apcahe的HTTPClient工具Http请求当请求超时重发
		
    10. 
						
  10. DLL入门浅析(2)——如何使用DLL
		
    11. 
						
  11. 编程实战——电影管理器之界面UI及动画切换
		
    12. 
						
  12. 关于在vs中添加生成命令时的注意事项
		
    13. 
						
  13. Compare and Swap [CAS] 算法
		
    14. 
						
  14. c# 网页打印全流程
		
    15. 
						
  15. 在做自动化测试之前你需要知道的，转自：http://www.cnblogs.com/fnng/p/3653793.html
		
    16. 
						
  16. GXB动态重配置
		
    17. 
						
  17. Windows+Linux双系统所导致的时间不一致问题的原因及解决方案
		
    18. 
						
  18. django基础 --  10.form , ModelForm ,modelformset
		
    19. 
						
  19. HTML语言和CSS开发
		
    20. 
						
  20. 基于Dlib、OpenCV开发人脸识别程序的开发建议
		
    21. 
		

	


热门文章
	

    
									
  1. no matches for kind "ReplicaSet" in version "extensions/v1beta1"
			
    2. 
						
  2. Hexo + github pages + 阿里云绑定域名搭建个人博客
			
    3. 
						
  3. jpa repostiory
			
    4. 
						
  4. MVC 小demo
			
    5. 
						
  5. Java——利用反射机制将表单数据自动填充到JavaBean中
			
    6. 
						
  6. spark 笔记 1: 如何着手
			
    7. 
						
  7. Eclipse如何安装Fat Jar
			
    8. 
						
  8. Vs code工具汉化
			
    9. 
						
  9. JavaEE-实验三 Java数据库高级编程
			
    10. 
						
  10. Cinder LVM backend cinder-volume service down
			
    11.