云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

日志服务 消费 Demo-多语言 SDK

• 腾讯云产品

这篇说说使用 Python、Java、Go SDK 通过 Kafka 协议消费腾讯云 CLS 日志，包括单个/多个消费者示例及配置注意事项。 Python SDK：推荐 Python 3.9+，使用 kafka-python 客户端，需配置主题、消费组、服务地址及 SASL 认证。 Java SDK：使用 kafka-clients 2.5.0，jaas.config 中密码后需加分号，配置示例包含完整消费者参数。 。

相关产品
日志服务 产品页，含版本规格、价格与购买入口。

这篇说说使用 Python、Java 消费 CLS 日志。

一、Python SDK

注意：说明：

二、单个消费者

import uuidfrom kafka import KafkaConsumer,TopicPartition,OffsetAndMetadataconsumer = KafkaConsumer(# cls kafka 协议消费控制台给出的的主题名称，例如XXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX，可在控制台复制'你的消费主题', group_id = '你的消费组名称',auto_offset_reset='earliest',# 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据你的实际情况填写bootstrap_servers = ['kafkaconsumer-${region}.cls.tencentyun.com:9095'],security_protocol = "SASL_PLAINTEXT",sasl_mechanism = 'PLAIN', # 用户名是日志集合ID，例如ca5cXXXXdd2e-4ac0af12-92d4b677d2c6 sasl_plain_username = "${logsetID}",# 密码是用户的 SecretId#SecretKey 组合的字符串，例如 AKID********************************#XXXXuXtymIXT0Lac注意不要丢失#。建议使用子账号密钥，主账号为子账号授权时，遵循最小权限原则，即子账号的访问策略中的 action、resource 都配置为最小范围，可以满足操作即可。sasl_plain_password = "${SecretId}#${SecretKey}",api_version = (0,10,1))print('begin')for message in consumer: print('begins') print ("Topic:[%s] Partition:[%d] Offset:[%d] Value:[%s]" % (message.topic, message.partition, message.offset, message.value)) print('end')

三、多个消费者

from kafka import KafkaConsumerimport threadingTOPIC_NAME = '你的消费主题'GROUP_ID = '你的消费组名称'# 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据你的实际情况填写BOOTSTRAP_SERVERS = ''kafkaconsumer-${region}.cls.tencentyun.com:9095''def consume_messages(thread_id): # 创建 Kafka 消费者实例 consumer = KafkaConsumer( TOPIC_NAME, group_id=GROUP_ID, bootstrap_servers=BOOTSTRAP_SERVERS, value_deserializer=lambda m: m.decode('utf-8'), auto_offset_reset='earliest', security_protocol = "SASL_PLAINTEXT", sasl_mechanism = 'PLAIN', sasl_plain_username = "${logsetID}", sasl_plain_password = "${SecretId}#${SecretKey}", api_version = (2, 5, 1) ) try: for message in consumer: print(f"Thread {thread_id}: partition = {message.partition}, offset = {message.offset}, value = {message.value}") except KeyboardInterrupt: pass finally: # 关闭消费者 consumer.close()if __name__ == "__main__": # 启动3个消费者线程，这是个例子，请你根据实际情况配置 num_consumers = 3 threads = [] for i in range(num_consumers): thread = threading.Thread(target=consume_messages, args=(i,)) threads.append(thread) thread.start() # 等待所有线程结束 for thread in threads: thread.join()

四、Java SDK

注意：下面的例子中的 Java 代码，在 jaas.config 的配置中，${SecretId}#${SecretKey} 后有（; 分号），不要漏填，否则会报错。

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.5.0</version></dependency>import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;import java.util.Collections;import java.util.Properties;public class KafkaConsumerGroupTest { public static void consume() { Properties props = new Properties(); String logset_id = "${logsetID}"; // CLS 控制台 kafka 协议消费页面展示的主题名称 String topic_id = "你的消费主题"; String accessKeyID = System.getenv("${SecretId}"); String accessKeySecret = System.getenv("${SecretKey}"); String groupId = "你的消费组名称"; // 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据你的实际情况填写 String hosts = "kafkaconsumer-${region}.cls.tencentyun.com:9095"; props.put("bootstrap.servers", hosts); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "PLAIN"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\\"" + logset_id + "\\" password=\\"" + accessKeyID + "#" + accessKeySecret + "\\";"); // Kafka 消费者配置 props.put("group.id", groupId); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "5000"); props.put("session.timeout.ms", "10000"); props.put("auto.offset.reset", "earliest"); props.put("max.poll.interval.ms", "120000"); props.put("heartbeat.interval.ms", "3000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 创建 Kafka 消费者实例 KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(props); consumer.subscribe(Collections.singletonList(topic_id)); while(true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10000)); for (ConsumerRecord<String, String> record : records) { System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); } } } public static void main(String[] args){ consume(); }}

五、Go SDK

注意：下面的例子中的代码，在 jaas.config 的配置中，${SecretId}#${SecretKey} 后有（; 分号），不要漏填，否则会报错。

package mainimport ( "context" "fmt" "github.com/Shopify/sarama" "log" "os" "os/signal" "syscall")func main() { // 创建Sarama消费者配置 //TOPIC_NAME是你的消费主题,在控制台查看. topicName := "${TOPIC_NAME}" //GROUP_ID 是 你的消费组名称 groupID := "${GROUP_ID}" //BOOTSTRAP_SERVERS 是消费服务地址+端口，外网端口9096，内网端口9095，例如 kafkaconsumer-${region}.cls.tencentyun.com:9095 endpoint := "${BOOTSTRAP_SERVERS}" config := sarama.NewConfig() config.Net.SASL.Enable = true config.Net.SASL.User = "${logsetID}" config.Net.SASL.Password = "${SecretId}#${SecretKey}" config.Net.SASL.Mechanism = sarama.SASLTypePlaintext config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin config.Version = sarama.V1_1_1_0 config.Consumer.Offsets.Initial = sarama.OffsetNewest // 创建Sarama消费者 sarama.Logger = log.New(os.Stdout, "[Sarama] ", log.LstdFlags) consumer, err := sarama.NewConsumerGroup([]string{endpoint}, groupID, config) if err != nil { log.Fatal(err) } defer consumer.Close() // 处理接收到的消息 handler := &ConsumerGroupHandler{} signals := make(chan os.Signal, 1) signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) go func() { for { err := consumer.Consume(context.Background(), []string{topicName}, handler) if err != nil { log.Fatal(err) } if handler.ready { break } } }() <-signals fmt.Println("Exiting…")}// ConsumerGroupHandler 实现了sarama.ConsumerGroupHandler接口type ConsumerGroupHandler struct { ready bool}// Setup 在消费者组启动之前调用func (h *ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error { h.ready = true return nil}// Cleanup 在消费者组停止之后调用func (h *ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error { h.ready = false return nil}// ConsumeClaim 消费Claim中的消息func (h *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { for message := range claim.Messages() { fmt.Printf("Received message: %s\\n", string(message.Value)) session.MarkMessage(message, "") } return nil}

相关内容：

  1. 日志服务 消费 Demo-流处理 这篇说说使用 Flink 和腾讯云 Oceanus 消费 CLS 日志的两种方式，包括通过 Oceanus SQL 作业和原生 Flink 注册 Kafka 表，均需配置 SASL 认证与消费组信息。 Oceanus 消费：在 Oceanus 控制台新建 SQL 作业，编写含 SASL 认证的 Kafka 表 SQL 语句。 Flink 消费：开启 CLS 日志的 Kafka 消费协议，在 Flink SQL 中注册。...
  2. 日志服务 消费 Demo-第三方软件 使用 Filebeat、 Logstash、Flume 消费日志，以及通过 HTTP 方式将你消费的日志发送至 SIEM（Splunk、Devo）。...
  3. 消息队列 CKafka 版 Schema Registry 接入 CKafka 无论是使用传统的 Avro API 自定义序列化类与反序列化类，还是使用 Twitter 的 Bijection 类库实现 Avro 的序列化与反序列化，两种方法有相同的缺点：在每条 Kafka 记录里都嵌入了 Schema，从而导致记录的大小成倍增加。但是不管怎样，在读取记录时仍然需要用到整个 Schema，所以要先找到 Schema。 CKafka 提供了数据共用一个 Schema 的方法：将 Schema 中的。...
  4. 消息队列 CKafka 版 自动重置消费进度 这篇说说Kafka参数auto.offset.reset，定义当无法获取消费分区位移时从何处开始消费，可选earliest、latest（默认）和none，并说明OFFSET_OUT_OF_RANGE错误原因及none模式的使用方式。 参数选项：earliest重置到最小offset，latest到最大offset，none不自动重置。 OFFSET_OUT_OF_RANGE：因日志过期、SDK异常或leader切换。...
  5. 日志服务 使用 Kafka 协议上传日志 日志服务（Cloud Log Service，CLS）目前已支持使用 Kafka Producer SDK 和其他 Kafka 相关 agent 上传日志到 CLS。...
  6. 消息队列 CKafka 版 配置事务消息 Kafka 的事务功能是为了支持在分布式环境中实现原子性操作而设计的。它允许生产者在发送消息时确保消息的完整性和一致性，特别是在需要多条消息作为一个整体进行处理的场景中。以下是 Kafka 事务的主要概念和功能介绍。...
  7. 消息队列 CKafka 版 接入低版本自建 Kafka CKafka 兼容0.9及以上的生产/消费接口（目前可以直接购买的版本包括 2.4.1、2.8.1、3.2.3 版本），如果接入低版本（例如0.8版本）的自建 Kafka，需要对接口进行相应改造。这篇将从生产端和消费端对比0.8版本 Kafka 和高版本 Kafka，并提供改造方式。...
  8. 日志服务 消费 Demo-消费预过滤 这篇说说如何使用Python SDK实现腾讯云日志服务（CLS）的数据预过滤与消费，通过配置query参数可在服务端过滤日志后再消费，节省带宽，并给出消费者类实现示例及消费任务启动配置。 预过滤：配置query参数实现，不配置则全量消费。 消费者实现：继承ConsumerProcessorBase类，在process方法中处理日志数据。 并发建议：消费者数量应与日志主题分区数保持一致。...
赞 (0)
上一篇 4天前
下一篇 4天前