java(kotlin)和 python 通过DoubleCloud的kafka举行线程间通信

进入
DoubleCloud
https://www.double.cloud
创建一个kafka
1 选择语言
2 运行curl 的url下令启动一个topic
3 生成对应语言的token
4 复制3中的配置文件到当地，命名为client.properties
5 复制客户端代码
对python和java客户端代码举行了重写，java改成了kotlin：
配置文件

1. # Required connection configs for Kafka producer, consumer, and admin
2. bootstrap.servers=
3. security.protocol=SASL_SSL
4. sasl.mechanisms=PLAIN
5. sasl.username=
6. sasl.password=
7. group.id=
8. auto.offset.reset=earliest
9. # Best practice for higher availability in librdkafka clients prior to 1.7
10. session.timeout.ms=45000

复制代码

1. import time
3. from confluent_kafka import Producer, Consumer
4. import asyncio
5. import threading
8. class KafkaClient:
9.     def __init__(self, config_file):
10.         self.config = self.read_config(config_file)
12.     def read_config(self, config_file):
13.         config = {}
14.         with open(config_file) as fh:
15.             for line in fh:
16.                 line = line.strip()
17.                 if len(line) != 0 and line[0] != "#":
18.                     parameter, value = line.strip().split('=', 1)
19.                     config[parameter] = value.strip()
20.         return config
22.     def produce(self, topic, key, value):
23.         # Creates a new producer instance
24.         producer = Producer(self.config)
26.         # Produces a sample message
27.         producer.produce(topic, key=key, value=value)
28.         print(f"Produced message to topic {topic}: key = {key:12} value = {value:12}")
30.         # Send any outstanding or buffered messages to the Kafka broker
31.         producer.flush()
33.     def consume_async(self, topic, callback=None, group_id="python-group-1", auto_offset_reset="earliest"):
34.         # Sets the consumer group ID and offset
35.         self.config["group.id"] = group_id
36.         self.config["auto.offset.reset"] = auto_offset_reset
37.         consumer = Consumer(self.config)
38.         consumer.subscribe([topic])
39.         loop = asyncio.new_event_loop()
40.         asyncio.set_event_loop(loop)
41.         if callback is not None:
42.             loop.run_until_complete(callback(consumer))
44.     def consume(self, topic, callback=None):
45.         thread = threading.Thread(target=self.consume_async, args=(topic, callback,))
46.         thread.start()
47.         return thread
50. async def consume_async(consumer):
51.     try:
52.         while True:
53.             msg = consumer.poll(1.0)
54.             if msg is not None:
55.                 break
56.         if not msg.error():
57.             key = msg.key().decode("utf-8")
58.             value = msg.value().decode("utf-8")
59.             print(f"Consumed message: key = {key:12} value = {value:12}")
60.     except KeyboardInterrupt:
61.         pass
62.     finally:
63.         consumer.close()
66. config_file_path = ".\\client.properties"
67. topic = "test"
68. key = "key"
69. value = "value"
71. kafka_client = KafkaClient(config_file_path)
72. kafka_client.produce(topic, key, value)
73. thread = kafka_client.consume(topic, consume_async)

复制代码

配置文件

1. # Required connection configs for Kafka producer, consumer, and admin
2. bootstrap.servers=
3. security.protocol=SASL_SSL
4. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='GHFXZDIOMQW3IPKA' password='TimUk7hj/EwTiB031lA95LeKfXN3t2Ddnw+izhKx3+7wFxZKMLGEqTOnneTKrlQQ';
5. sasl.mechanism=PLAIN
6. # Required for correctness in Apache Kafka clients prior to 2.6
7. client.dns.lookup=use_all_dns_ips
8. key.serializer=org.apache.kafka.common.serialization.StringSerializer
9. value.serializer=org.apache.kafka.common.serialization.StringSerializer
10. # Best practice for higher availability in Apache Kafka clients prior to 3.0
11. session.timeout.ms=45000
12. topic=
13. group.id=
14. auto.offset.reset=earliest
15. key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
16. value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
17. # Best practice for Kafka producer to prevent data loss
18. acks=all

复制代码

java(kotiln)

2. import kotlinx.coroutines.CoroutineScope
3. import kotlinx.coroutines.delay
4. import kotlinx.coroutines.launch
5. import kotlinx.coroutines.newFixedThreadPoolContext
6. import org.apache.kafka.clients.consumer.ConsumerRecords
7. import org.apache.kafka.clients.consumer.KafkaConsumer
8. import org.apache.kafka.clients.producer.KafkaProducer
9. import org.apache.kafka.clients.producer.ProducerRecord
10. import java.io.Closeable
11. import java.io.FileInputStream
12. import java.io.IOException
13. import java.nio.file.Files
14. import java.nio.file.Paths
15. import java.time.Duration
16. import java.util.*
18. class KafkaClient<T, V> : Closeable {
19.     private var producer: KafkaProducer<T, V>? = null
20.     private var fileConfig: Properties? = null
21.     val TOPIC = "topic"
22.     private val DURATION = 100L
23.     private val POOLSIZE = 10
24.     private val DISPATCHER = newFixedThreadPoolContext(POOLSIZE, "CoroutinePool")
25.     private val SCOPE = CoroutineScope(DISPATCHER)
27.     constructor(configPath: String? = null, config: Properties? = null) {
28.         if (config == null && configPath == null) throw Exception("don't have any config")
29.         var config1 = Properties()
30.         if (configPath != null) {
31.             fileConfig = readConfig(configPath)
32.             fileConfig?.let { config1.putAll(it) }
33.         }
34.         if (config != null) {
35.             config1.putAll(config)
36.         }
37.         producer = KafkaProducer(config1)
38.     }
40.     fun produce(key: T, value: V, topic: String? = null) {
41.         producer?.send(ProducerRecord(topic ?: (fileConfig?.getProperty(TOPIC) as String), key, value))
42.     }
44.     fun consume(func: suspend (ConsumerRecords<T, V>) -> Unit) {
45.         val consumer: KafkaConsumer<T, V> = KafkaConsumer(fileConfig)
46.         consumer.subscribe(Arrays.asList(fileConfig?.getProperty(TOPIC)))
47.         SCOPE.launch {
48.             while (true) {
49.                 val records: ConsumerRecords<T, V> = consumer.poll(Duration.ofMillis(DURATION))
50.                 func(records)
51.                 delay(DURATION)
52.             }
53.         }
54.     }
57.     @Throws(IOException::class)
58.     fun readConfig(configFile: String): Properties {
59.         if (!Files.exists(Paths.get(configFile))) {
60.             throw IOException("$configFile not found.")
61.         }
62.         val config = Properties()
63.         FileInputStream(configFile).use { inputStream -> config.load(inputStream) }
64.         return config
65.     }
67.     override fun close() {
68.         producer?.close()
69.     }
70. }
72. fun main() {
73.     val cli =
74.         KafkaClient<String, String>("D:\\src\\main\\java\\com\\tr\\robot\\io\\kafka\\client.properties")
75.     cli.consume {
76.         println("test beg")
77.         for (record in it) {
78.             println(
79.                 String.format(
80.                     "Consumed message from topic %s: key = %s value = %s", cli.TOPIC, record.key(), record.value()
81.                 )
82.             )
83.         }
84.         println("test end")
85.     }
87.     // Give some time for the consumer to start
88.     Thread.sleep(2000)
90.     cli.produce("key1", "test")
92.     // Give some time for the consumer to consume the message
93.     Thread.sleep(5000)
94. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。