Go操作Kafka之kafka-go

Kafka是一种高吞吐量的分布式发布订阅消息系统，本文介绍了如何使用kafka-go这个库实现Go语言与kafka的交互。

Go社区中目前有三个比较常用的kafka客户端库 , 它们各有特点。

首先是IBM/sarama（这个库已经由Shopify转给了IBM），之前我写过一篇使用sarama操作Kafka的教程，相较于sarama， kafka-go 更简单、更易用。

segmentio/kafka-go 是纯Go实现，提供了与kafka交互的低级别和高级别两套API，同时也支持Context。

此外社区中另一个比较常用的confluentinc/confluent-kafka-go，它是一个基于cgo的librdkafka包装，在项目中使用它会引入对C库的依赖。

准备Kafka环境

这里推荐使用Docker Compose快速搭建一套本地开发环境。

以下docker-compose.yml文件用来搭建一套单节点zookeeper和单节点kafka环境，并且在8080端口提供kafka-ui管理界面。

version: '2.1'

services:
  zoo1:
    image: confluentinc/cp-zookeeper:7.3.2
    hostname: zoo1
    container_name: zoo1
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_SERVERS: zoo1:2888:3888

  kafka1:
    image: confluentinc/cp-kafka:7.3.2
    hostname: kafka1
    container_name: kafka1
    ports:
      - "9092:9092"
      - "29092:29092"
      - "9999:9999"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092,DOCKER://host.docker.internal:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_JMX_PORT: 9999
      KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1}
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    depends_on:
      - zoo1
  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    ports:
      - 8080:8080
    depends_on:
      - kafka1
    environment:
      DYNAMIC_CONFIG_ENABLED: "TRUE"

将上述docker-compose.yml文件在本地保存，在同一目录下执行以下命令启动容器。

docker-compose up -d

容器启动后，使用浏览器打开127.0.0.1:8080 即可看到如下kafka-ui界面。

在这里插入图片描述

点击页面右侧的“Configure new cluster”按钮，配置kafka服务连接信息。

在这里插入图片描述
填写完信息后，点击页面下方的“Submit”按钮提交即可。

在这里插入图片描述

安装kafka-go

执行以下命令下载 kafka-go依赖。

go get github.com/segmentio/kafka-go

注意：kafka-go 需要 Go 1.15或更高版本。

kafka-go使用指南

kafka-go 提供了两套与Kafka交互的API。

  • 低级别（ low-level）：基于与 Kafka 服务器的原始网络连接实现。
  • 高级别（high-level）：对于常用读写操作封装了一套更易用的API。

通常建议直接使用高级别的交互API。

Connection

Conn 类型是 kafka-go 包的核心。它代表与 Kafka broker之间的连接。基于它实现了一套与Kafka交互的低级别 API。

发送消息

下面是连接至Kafka之后，使用Conn发送消息的代码示例。

// writeByConn 基于Conn发送消息
func writeByConn() {
	topic := "my-topic"
	partition := 0

	// 连接至Kafka集群的Leader节点
	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	// 设置发送消息的超时时间
	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

	// 发送消息
	_, err = conn.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	// 关闭连接
	if err := conn.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
}
消费消息
// readByConn 连接至kafka后接收消息
func readByConn() {
	// 指定要连接的topic和partition
	topic := "my-topic"
	partition := 0

	// 连接至Kafka的leader节点
	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	// 设置读取超时时间
	conn.SetReadDeadline(time.Now().Add(10 * time.Second))
	// 读取一批消息，得到的batch是一系列消息的迭代器
	batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

	// 遍历读取消息
	b := make([]byte, 10e3) // 10KB max per message
	for {
		n, err := batch.Read(b)
		if err != nil {
			break
		}
		fmt.Println(string(b[:n]))
	}

	// 关闭batch
	if err := batch.Close(); err != nil {
		log.Fatal("failed to close batch:", err)
	}

	// 关闭连接
	if err := conn.Close(); err != nil {
		log.Fatal("failed to close connection:", err)
	}
}

使用batch.Read更高效一些，但是需要根据消息长度选择合适的buffer（上述代码中的b），如果传入的buffer太小（消息装不下）就会返回io.ErrShortBuffer错误。

如果不考虑内存分配的效率问题，也可以按以下代码使用batch.ReadMessage读取消息。

for {
  msg, err := batch.ReadMessage()
  if err != nil {
    break
  }
  fmt.Println(string(msg.Value))
}
创建topic

当Kafka关闭自动创建topic的设置时，可按如下方式创建topic。

// createTopicByConn 创建topic
func createTopicByConn() {
	// 指定要创建的topic名称
	topic := "my-topic"

	// 连接至任意kafka节点
	conn, err := kafka.Dial("tcp", "localhost:9092")
	if err != nil {
		panic(err.Error())
	}
	defer conn.Close()

	// 获取当前控制节点信息
	controller, err := conn.Controller()
	if err != nil {
		panic(err.Error())
	}
	var controllerConn *kafka.Conn
	// 连接至leader节点
	controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
	if err != nil {
		panic(err.Error())
	}
	defer controllerConn.Close()

	topicConfigs := []kafka.TopicConfig{
		{
			Topic:             topic,
			NumPartitions:     1,
			ReplicationFactor: 1,
		},
	}

	// 创建topic
	err = controllerConn.CreateTopics(topicConfigs...)
	if err != nil {
		panic(err.Error())
	}
}
通过非leader节点连接leader节点

下面的示例代码演示了如何通过已有的非leader节点的Conn，连接至 leader节点。

conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()
// 获取当前控制节点信息
controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}
var connLeader *kafka.Conn
connLeader, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer connLeader.Close()
获取topic列表
conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

partitions, err := conn.ReadPartitions()
if err != nil {
    panic(err.Error())
}

m := map[string]struct{}{}
// 遍历所有分区取topic
for _, p := range partitions {
    m[p.Topic] = struct{}{}
}
for k := range m {
    fmt.Println(k)
}

Reader

Reader是由 kafka-go 包提供的另一个概念，对于从单个主题-分区（topic-partition）消费消息这种典型场景，使用它能够简化代码。Reader 还实现了自动重连和偏移量管理，并支持使用 Context 支持异步取消和超时的 API。

注意： 当进程退出时，必须在 Reader 上调用 Close() 。Kafka服务器需要一个优雅的断开连接来阻止它继续尝试向已连接的客户端发送消息。如果进程使用 SIGINT (shell 中的 Ctrl-C)或 SIGTERM (如 docker stop 或 kubernetes start)终止，那么下面给出的示例不会调用 Close()。当同一topic上有新Reader连接时，可能导致延迟(例如，新进程启动或新容器运行)。在这种场景下应使用signal.Notify处理程序在进程关闭时关闭Reader。

消费消息

下面的代码演示了如何使用Reader连接至Kafka消费消息。

// readByReader 通过Reader接收消息
func readByReader() {
	// 创建Reader
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{"localhost:9092", "localhost:9093", "localhost:9094"},
		Topic:     "topic-A",
		Partition: 0,
		MaxBytes:  10e6, // 10MB
	})
	r.SetOffset(42) // 设置Offset

	// 接收消息
	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			break
		}
		fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
	}

	// 程序退出前关闭Reader
	if err := r.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}
}
消费者组

kafka-go支持消费者组，包括broker管理的offset。要启用消费者组，只需在 ReaderConfig 中指定 GroupID。

使用消费者组时，ReadMessage 会自动提交偏移量。

// 创建一个reader，指定GroupID，从 topic-A 消费消息
r := kafka.NewReader(kafka.ReaderConfig{
	Brokers:  []string{"localhost:9092", "localhost:9093", "localhost:9094"},
	GroupID:  "consumer-group-id", // 指定消费者组id
	Topic:    "topic-A",
	MaxBytes: 10e6, // 10MB
})

// 接收消息
for {
	m, err := r.ReadMessage(context.Background())
	if err != nil {
		break
	}
	fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
}

// 程序退出前关闭Reader
if err := r.Close(); err != nil {
	log.Fatal("failed to close reader:", err)
}

在使用消费者组时会有以下限制：

(*Reader).SetOffset 当设置了GroupID时会返回错误
(*Reader).Offset 当设置了GroupID时会永远返回 -1
(*Reader).Lag 当设置了GroupID时会永远返回 -1
(*Reader).ReadLag 当设置了GroupID时会返回错误
(*Reader).Stats 当设置了GroupID时会返回一个-1的分区

显式提交

kafka-go 也支持显式提交。当需要显式提交时不要调用 ReadMessage，而是调用 FetchMessage获取消息，然后调用 CommitMessages 显式提交。

ctx := context.Background()
for {
    // 获取消息
    m, err := r.FetchMessage(ctx)
    if err != nil {
        break
    }
    // 处理消息
    fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
    // 显式提交
    if err := r.CommitMessages(ctx, m); err != nil {
        log.Fatal("failed to commit messages:", err)
    }
}

在消费者组中提交消息时，具有给定主题/分区的最大偏移量的消息确定该分区的提交偏移量的值。例如，如果通过调用 FetchMessage 获取了单个分区的偏移量为 1、2 和 3 的消息，则使用偏移量为3的消息调用 CommitMessages 也将导致该分区的偏移量为 1 和 2 的消息被提交。

管理提交间隔

默认情况下，调用CommitMessages将同步向Kafka提交偏移量。为了提高性能，可以在ReaderConfig中设置CommitInterval来定期向Kafka提交偏移。

// 创建一个reader从 topic-A 消费消息
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092", "localhost:9093", "localhost:9094"},
    GroupID:        "consumer-group-id",
    Topic:          "topic-A",
    MaxBytes:       10e6, // 10MB
    CommitInterval: time.Second, // 每秒刷新一次提交给 Kafka
})

Writer

向Kafka发送消息，除了使用基于Conn的低级API，kafka-go包还提供了更高级别的 Writer 类型。大多数情况下使用Writer即可满足条件，它支持以下特性。

  • 对错误进行自动重试和重新连接。
  • 在可用分区之间可配置的消息分布。
  • 向Kafka同步或异步写入消息。
  • 使用Context的异步取消。
  • 关闭时清除挂起的消息以支持正常关闭。
  • 在发布消息之前自动创建不存在的topic。
发送消息
// 创建一个writer 向topic-A发送消息
w := &kafka.Writer{
	Addr:         kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
	Topic:        "topic-A",
	Balancer:     &kafka.LeastBytes{}, // 指定分区的balancer模式为最小字节分布
	RequiredAcks: kafka.RequireAll,    // ack模式
	Async:        true,                // 异步
}

err := w.WriteMessages(context.Background(),
	kafka.Message{
		Key:   []byte("Key-A"),
		Value: []byte("Hello World!"),
	},
	kafka.Message{
		Key:   []byte("Key-B"),
		Value: []byte("One!"),
	},
	kafka.Message{
		Key:   []byte("Key-C"),
		Value: []byte("Two!"),
	},
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

if err := w.Close(); err != nil {
    log.Fatal("failed to close writer:", err)
}
创建不存在的topic

如果给Writer配置了AllowAutoTopicCreation:true，那么当发送消息至某个不存在的topic时，则会自动创建topic。


// 创建不存在的topic
// 如果给Writer配置了AllowAutoTopicCreation:true，那么当发送消息至某个不存在的topic时，则会自动创建topic。
func writeByWriter2() {
	writer := kafka.Writer{
		Addr:                   kafka.TCP("192.168.2.204:9092"),
		Topic:                  "kafka-test-topic",
		AllowAutoTopicCreation: true, //自动创建topic
	}

	messages := []kafka.Message{
		{
			Key:   []byte("Key-A"),
			Value: []byte("Hello World!"),
		},
		{
			Key:   []byte("Key-B"),
			Value: []byte("One!"),
		},
		{
			Key:   []byte("Key-C"),
			Value: []byte("Tow!"),
		},
	}

	const retries = 3
	//重试3次
	for i := 0; i < retries; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()

		err := writer.WriteMessages(ctx, messages...)
		if errors.Is(err, kafka.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
			time.Sleep(time.Millisecond * 250)
			continue
		}

		if err != nil {
			log.Fatal("unexpected error %v", err)
		}
		break
	}

	//关闭Writer
	if err := writer.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
}
写入多个topic

通常，WriterConfig.Topic用于初始化单个topic的Writer。通过去掉WriterConfig中的Topic配置，分别设置每条消息的message.topic，可以实现将消息发送至多个topic。

w := &kafka.Writer{
	Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    // 注意: 当此处不设置Topic时,后续的每条消息都需要指定Topic
	Balancer: &kafka.LeastBytes{},
}

err := w.WriteMessages(context.Background(),
    // 注意: 每条消息都需要指定一个 Topic, 否则就会报错
	kafka.Message{
        Topic: "topic-A",
		Key:   []byte("Key-A"),
		Value: []byte("Hello World!"),
	},
	kafka.Message{
        Topic: "topic-B",
		Key:   []byte("Key-B"),
		Value: []byte("One!"),
	},
	kafka.Message{
        Topic: "topic-C",
		Key:   []byte("Key-C"),
		Value: []byte("Two!"),
	},
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

if err := w.Close(); err != nil {
    log.Fatal("failed to close writer:", err)
}

注意：Writer中的Topic和Message中的Topic是互斥的，同一时刻有且只能设置一处。

其他配置

TLS

对于基本的 Conn 类型或在 Reader/Writer 配置中，可以在Dialer中设置TLS选项。如果 TLS 字段为空，则它将不启用TLS 连接。

注意：不在Conn/Reder/Writer上配置TLS，连接到启用TLS的Kafka集群，可能会出现io.ErrUnexpectedEOF错误。

Connection
dialer := &kafka.Dialer{
    Timeout:   10 * time.Second,
    DualStack: true,
    TLS:       &tls.Config{...tls config...},  // 指定TLS配置
}

conn, err := dialer.DialContext(ctx, "tcp", "localhost:9093")
Reader
dialer := &kafka.Dialer{
    Timeout:   10 * time.Second,
    DualStack: true,
    TLS:       &tls.Config{...tls config...},  // 指定TLS配置
}

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092", "localhost:9093", "localhost:9094"},
    GroupID:        "consumer-group-id",
    Topic:          "topic-A",
    Dialer:         dialer,
})
Writer

创建Writer时可以按如下方式指定TLS配置。

w := kafka.Writer{
    Addr: kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"), 
    Topic:   "topic-A",
    Balancer: &kafka.Hash{},
    Transport: &kafka.Transport{
        TLS: &tls.Config{},  // 指定TLS配置
      },
    }

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/540421.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

WebGL 2.0相较于1.0有什么不同？

WebGL 2.0相较于1.0有什么不同？

作者&#xff1a;STANCH 1.概述 WebGL 1.0自推出以来&#xff0c;已成为广泛支持的Web标准&#xff0c;既能跨平台&#xff0c;还免版税。它通过插件为Web浏览器带来高质量的3D图形&#xff0c;这是迄今为止市场上使用最广泛的Web图形&#xff0c;并得到Apple&#xff0c;Goog…
阅读更多...
Arduino _按键点亮——led

Arduino _按键点亮——led

int8_t led_pin13;int8_t led2_pin12; void setup() {// put your setup code here, to run once:pinMode(led_pin, INPUT);//输入pinMode(led2_pin, OUTPUT);//输出 }void loop() {// put your main code here, to run repeatedly:if(digitalRead(led_pin)1){//digitalRead(…
阅读更多...
智慧校园平台解决方案-迎新管理系统

智慧校园平台解决方案-迎新管理系统

数字迎新管理系统是整个智慧校园不可缺少的一部分&#xff1b;主要是为了方便学校优化迎新流程&#xff0c;规范迎新流程&#xff0c;高效率的迎新&#xff0c;减少学生迎新繁琐流程。现阶段自友科技数字迎新利用了网络方式在网上即可提前知道迎新流程&#xff0c;学生家长在手…
阅读更多...
【python】python饮料销售数据分析可视化（源码+数据集）【独一无二】

【python】python饮料销售数据分析可视化（源码+数据集）【独一无二】

&#x1f449;博__主&#x1f448;&#xff1a;米码收割机 &#x1f449;技__能&#x1f448;&#xff1a;C/Python语言 &#x1f449;公众号&#x1f448;&#xff1a;测试开发自动化【获取源码商业合作】 &#x1f449;荣__誉&#x1f448;&#xff1a;阿里云博客专家博主、5…
阅读更多...
高效保护，无粉乳胶手套助您安心工作

高效保护，无粉乳胶手套助您安心工作

在快节奏的现代生活中&#xff0c;我们每天都在与各种细菌、污染物和化学品打交道&#xff0c;因此保护我们的身体健康变得愈发重要。特别是在工作环境中&#xff0c;手部是最容易受到伤害和污染的部位之一。为了提供最佳的保护&#xff0c;优斯特推出了一款高品质的无粉乳胶手…
阅读更多...
Web3 的社会影响：数字社会的新时代

Web3 的社会影响：数字社会的新时代

随着科技的不断进步和创新&#xff0c;人类社会正逐步进入数字化时代的新阶段。Web3 技术作为数字社会的重要组成部分&#xff0c;正在以前所未有的方式重塑着我们的社会生活和交往方式。本文将探讨 Web3 技术对社会的影响&#xff0c;以及它所带来的数字社会的新时代。 1. Web…
阅读更多...
阻塞队列和生产消费模型

阻塞队列和生产消费模型

阻塞队列 阻塞队列的概念 队列相信我们已经不陌生了 之前也学过很多队列 比如: 普通队列 和 优先级队列 两种 这两种队列都是线程不安全的 而我们讲的阻塞队列 刚好可以解决线程安全问题 也是先进先出 并且带有阻塞功能. 阻塞功能是怎么回事呢 就是如果入队的时候阻塞队列为…
阅读更多...
深入解析API技术：原理、实现与应用

深入解析API技术：原理、实现与应用

在现代软件开发中&#xff0c;API&#xff08;应用程序接口&#xff09;扮演着至关重要的角色。API 允许不同的软件应用程序和系统之间进行通信和数据交换&#xff0c;从而构建出更加高效、灵活和可扩展的软件解决方案。本文将深入解析API技术的原理、实现方法&#xff0c;并附…
阅读更多...
iptables/ebtables学习笔记

iptables/ebtables学习笔记

目录 一、前言 二、Netfilter 构成 三、Netfilter 转发框架 四、Netfilter 与 iptables 五、Netfilter 与 ebtables 一、前言 Netfilter 是 Linux 内核的数据包处理框架&#xff0c;由 Rusty Russell 于 1998 年开发&#xff0c; 旨在改进以前的 ipchains&#xff08;Lin…
阅读更多...
中霖教育：一级建造师能跨省考吗?

中霖教育：一级建造师能跨省考吗?

一级建造师是可以跨省考的&#xff0c;对于考生的户籍地并没有限制&#xff0c;只要符合参加考试的条件就可以报名。 但是&#xff0c;对于异地报名参加一级建造师考试的考生&#xff0c;必须经过严格的资料审核流程。考生在报名过程中需要提交居住证明、工作证明、以及异地考…
阅读更多...
c语言例题，求数组中最大值，99乘法口诀表

c语言例题，求数组中最大值，99乘法口诀表

例题1&#xff1a;求出数组中最大的值 根据题意&#xff0c;我们知道的是需要从一个数组中找到一个最大的元素并且输出。那首先我们先建立一个数组&#xff0c;然后将一些不有序的整型元素放到数组中&#xff0c;然后再建立一个变量来存放数组中的第一个元素&#xff0c;通过一…
阅读更多...
第23次修改了可删除可持久保存的前端html备忘录:增加了百度引擎

第23次修改了可删除可持久保存的前端html备忘录:增加了百度引擎

第22次修改了可删除可持久保存的前端html备忘录视频背景分离&#xff0c;增加了本地连接&#xff0c;增加了纯CSS做的折叠隐藏修改说明 <!DOCTYPE html> <html lang"zh"> <head><meta charset"UTF-8"><meta name"viewport…
阅读更多...
（三）ffmpeg 解码流程以及函数介绍

（三）ffmpeg 解码流程以及函数介绍

一、视频解码流程 二、函数介绍 1.avformat_network_init 函数作用&#xff1a; 执行网络库的全局初始化。这是可选的&#xff0c;不再推荐。 此函数仅用于解决旧GnuTLS或OpenSSL库的线程安全问题。如果libavformat链接到这些库的较新版本&#xff0c;或者不使用它们&#…
阅读更多...
跟TED演讲学英文：The dark side of competition in AI by Liv Boeree

跟TED演讲学英文：The dark side of competition in AI by Liv Boeree

The dark side of competition in AI Link: https://www.ted.com/talks/liv_boeree_the_dark_side_of_competition_in_ai Speaker:Liv Boeree Date: October 2023 文章目录 The dark side of competition in AIIntroductionVocabularyTranscriptSummary后记 Introduction Co…
阅读更多...
react 初学增删改查购物车案例

react 初学增删改查购物车案例

界面 代码 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8" /><meta name"viewport" content"widthdevice-width, initial-scale1.0" /><title>react-购物车案例</title><…
阅读更多...
[MQTT]Mosquitto的簡介、安裝與本機連接測試

[MQTT]Mosquitto的簡介、安裝與本機連接測試

MQTT(Message Queuing Telemetry Transport)是基於發布/訂閱模式(Publish/Subscribe)的輕量級網絡通訊協議&#xff0c;專用於在低帶寬、不可靠的網絡環境中傳輸數據&#xff0c;關於工作原理和優點如下: 1.發布/訂閱模式(Publish/Subscribe) 發布者將消息發布特定的主題(top…
阅读更多...
MIPI CSI-2 虚拟通道

MIPI CSI-2 虚拟通道

一、背景 在介绍虚拟通道之前&#xff0c;我们先描述一个场景&#xff0c;主控芯片只有 4 个 D-PHY VIN 接口&#xff0c;但是产品需要4 个以上的摄像头。 如上图所示&#xff0c;如果我们把摄像头和主控芯片直接通过 D-PHY 接口连接&#xff0c;最多只能接 4 个摄像头&#x…
阅读更多...
【Linux】基础IO----理解缓冲区

【Linux】基础IO----理解缓冲区

> 作者&#xff1a;დ旧言~ > 座右铭&#xff1a;松树千年终是朽&#xff0c;槿花一日自为荣。 > 目标&#xff1a;理解缓冲区 > 毒鸡汤&#xff1a;有些事情&#xff0c;总是不明白&#xff0c;所以我不会坚持。早安! > 专栏选自&#xff1a;Linux初阶 > 望…
阅读更多...
Linux 系统解压缩文件

Linux 系统解压缩文件

Linux系统&#xff0c;可以使用unzip命令来解压zip文件 方法如下 1. 打开终端&#xff0c;在命令行中输入以下命令来安装unzip&#xff1a; sudo apt-get install unzip 1 2. 假设你想要将zip文件解压缩到名为"target_dir"的目录中&#xff0c;在终端中切换到目标路…
阅读更多...
ActiveMQ介绍及linux下安装ActiveMQ

ActiveMQ介绍及linux下安装ActiveMQ

ActiveMQ介绍 概述 ActiveMQ是Apache软件基金下的一个开源软件&#xff0c;它遵循JMS1.1规范&#xff08;Java Message Service&#xff09;&#xff0c;是消息队列服务&#xff0c;是面向消息中间件&#xff08;MOM&#xff09;的最终实现&#xff0c;它为企业消息传递提供高…
阅读更多...
最新文章

Copyright @ 2022~2023