Database

redis basic commands redis-cli -h 127.0.0.1 -p 6379 # -a 使用认证密码登录 redis-cli -h 127.0.0.1 -p 6379 -a password0 # -n 指定 db redis-cli -h 127.0.0.1 -p 6379 -a password0 -n 10 # OBJECT ENCODING 命令可以查看一个数据库键的值对象的编码 OBJECT ENCODING key0 # 分析 redis key 大小 debug object key0 # Value at:0x7f6bffc22a00 refcount:1 encoding:raw serializedlength:7164 lru:12841785 lru_seconds_idle:95 # 查看各个库的 key 数量 info keyspace # 不进入交互模式, 直接执行命令 redis-cli -h 127.0.0.1 -p 6379 hget key0 field0 延迟时间 redis-cli --latency -h 192.168.50.100 -p 6379 sort https://segmentfault.com/a/1190000002806846 ...

kafka 分区数 Kafka的分区数是不是越多越好？ 分区多的优点 kafka使用分区将topic的消息打散到多个分区分布保存在不同的broker上，实现了producer和consumer消息处理的高吞吐量。Kafka的producer和consumer都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费。 所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。 分区不是越多越好 分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销: 一、客户端/服务器端需要使用的内存就越多 Kafka0.8.2之后，在客户端producer有个参数 batch.size，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。 服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本就越大。 二、文件句柄的开销 每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件: base_offset.log和base_offset.index。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。 三、降低高可用性 Kafka通过副本(replica)机制来保证高可用。具体做法就是为每个分区保存若干个副本(replica_factor指定副本数)。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。 如何确定分区数量呢？ 可以遵循一定的步骤来尝试确定分区数: 创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / max(Tp, Tc) 说明: Tp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大， 因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。 ———————————————— 版权声明: 本文为CSDN博主「AlferWei」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。 原文链接: https://blog.csdn.net/OiteBody/article/details/80595971

堆组织表、索引组织表、索引聚簇表 版权声明: 本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。 https://blog.csdn.net/pianzif/article/details/32106971 堆组织表就不说了，其索引中记录了记录所在位置的rowid，查找的时候先找索引，然后再根据索引rowid找到块中的行数据 索引组织表，其行数据以索引形式存放，因此找到索引，就等于找到了行数据。 堆组织表的数据是散放的，索引和表的数据是分离的 索引组织表的索引和数据是在一起的 堆组织表的存储速度因为不用考虑排序, 所以存储速度会比较快. 但是要查找符合某个条件的记录, 就必须得读取全部的记录以便筛选. 而这个时候为了加快查询速度, 索引就出现了, 索引是针对少量特定字段的值拿出来进行排序存储, 并记录在表中的位置, 而因为索引是有序的, 所以就会很容易通过索引查询到具体的记录位置, 然后再根据记录位置直接从表中读取该记录. 同时因为索引的字段较少, 所以索引通常会比其基表小得多. 从上面通过索引访问表记录的方式可以看出, 当要访问的数据量较大时, 通过每一条记录的位置去访问原始记录, 每一条符合条件的记录都需要经过索引访问后再访问基表这样一个复杂的过程, 这会花费很多时间, 同样, 如果不经过索引而直接查询表, 也可能因为表字段太多, 记录较大的情况下把全部的数据读取进来, 这也会花费很多时间. 那怎么办呢? 这个时候就会想到, 如果表中数据本身就是有序的, 这样查询表的时候就可以快速的找到符合条件的记录位置, 而很容易判断符合条件记录的位置, 这样只需要读取一小部分数据出来就可以了, 不需要全表记录都读取出来进行判断. 索引表就这样产生了.当然索引表中插入,更新资料的时候可能会因为需要排序而将数据重组, 这时候数据插入或更新速度会比堆组织表慢一些. 如果堆组织表上有索引, 那么对堆组织表的插入也会因为要修改索引而变慢 我们可以看到堆组织表+索引的方式 与 索引表 都能够实现数据的快速查找, 那为什么不全部采用索引表呢, 这样不是很简单吗? 我能想到的是前者我们可以针对不同的查找条件建立多个索引, 而后者却不行, 后者只能对某一组查询条件有效 堆组织表 (heap organized table) Oracle中有很多类型的表，像堆组织表、索引组织表、索引聚簇表等等。首先，我将从最基本、最常用的堆组织表 (heap organized table) 介绍。 通常我们默认建的表就是堆组织表。语法 (详细语法请参见Oracle官方文档) 如下: ...

数据库事务, 锁 数据库锁 因为数据库要解决并发控制问题。在同一时刻，可能会有多个客户端对同一张表进行操作，比如有的在读取该行数据，其他的尝试去删除它。为了保证数据的一致性，数据库就要对这种并发操作进行控制，因此就有了锁的概念。 锁的分类 从对数据库操作的类型分 读锁（共享锁）：针对同一块数据，多个读操作可以同时进行而不会互相影响。由读表操作加上的锁，加锁后其他用户只能获取该表或行的共享锁，不能获取排它锁，也就是说只能读不能写。 写锁（排它锁）：当当前写操作没有完成之前，它会阻断其他写锁和读锁。由写表操作加上的锁，加锁后其他用户不能获取该表或行的任何锁。 从锁定的数据范围分 表锁：锁定某个表。 行锁 ：锁定某行。 为了尽可能 提高数据库的并发度，每次锁定的数据范围越小越好。理论上每次只锁定当前操作的数据的方案会得到最大的并发度，但是管理锁是很耗费资源的事情。因此数据库系统需要在高并发响应和系统性能两方面进行平衡，这样就产生了“锁粒度”的概念。 锁粒度 表锁：管理锁的开销最小，同时允许的并发量也最小的锁机制。MyIsam存储引擎使用的锁机制。当要写入数据时，把整个表都锁上，此时其他读、写动作一律等待。在MySql中，除了MyIsam存储引擎使用这种锁策略外，MySql本身也使用表锁来执行某些特定动作，比如alter table. 行锁：可以支持最大并发的锁策略。InnoDB和Falcon两张存储引擎都采用这种策略。 MySql是一种开放的架构，你可以实现自己的存储引擎，并实现自己的锁粒度策略，不像Oracle，你没有机会改变锁策略，Oracle采用的是行锁。从大到小，mysql服务器仅支持表级锁，行锁需要存储引擎完成。粒度越精细，并发性越好。即行锁的并发性最好，但需要存储引擎的支持。 数据库事务有不同的隔离级别，不同的隔离级别对锁的使用是不同的，锁的应用最终导致不同事务的隔离级别。 事务和锁机制是什么关系？ 开启事务就自动加锁了吗？ 1、事务与锁是不同的。事务具有ACID（原子性、一致性、隔离性和持久性），锁是用于解决隔离性的一种机制。 2、事务的隔离级别通过锁的机制来实现。另外锁有不同的粒度，同时事务也是有不同的隔离级别的。 3、开启事务就自动加锁。 ql规范定义的事务的隔离级别： 1.READ UNCOMMITTED(读取未提交内容) 所有事务可以看到未提交事务的执行结果，本隔离级别很少用到实际应用中，读取未提交的数据，又称为“脏读”。 2.READ COMMITTED（读取提交内容） 大多数数据库的默认隔离级别是此级别，但不是MySQL默认的。一个事务在开始的时候只能看见已提交事务所做的改变。一个事务从开始到提交前所做的任何改变都是不可见的，除非提交。这种隔离级别也称为不可重复读。 3.REPEATABLE READ(可重复读) 此隔离级别是为了解决不可重复读隔离级别导致的问题即一个事务多个实例并发读取数据时会看到不同的结果。可重复读描述的是一个事务在处理数据时，多次查询此数据得到的结果是一样的。MySQL的InnoDB存储引擎通过多版本并发控制（Multi_Version Concurrency Control, MVCC）机制来解决该问题。不过理论上，这会导致另一个棘手的问题：幻读 （Phantom Read）。简单的说，幻读指当用户读取某一范围的数据行时， 另一个事务又在该范围内插入了新行，当用户再读取该范围的数据行时，会发现有新的“幻影” 行。（幻读描述的是MVCC不能阻止插入新的数据，导致多次查询时数据记录不一致。） 4.SERIALIZABLE(可串行化) 可串行化是最高的隔离级别，它通过强制事务排序，使之不可重读，解决了幻读的问题。此隔离级别会在每个读的数据行上加共享锁，使用这种隔离级别会产生大量的超时现象，一般实际开发中不会用到。该类型在A客户端操作test.test1表时会锁定该数据，如果B客户端想要操作test.test1就需要等待A客户端释放。 这四种隔离级别采取不同的锁类型来实现，若读取的是同一个数据的话，就容易发生问题。例如： 脏读(Drity Read)：某个事务已更新一份数据，另一个事务在此时读取了同一份数据，由于某些原因，前一个RollBack了操作，则后一个事务所读取的数据就会是不正确的。 不可重复读(Non-repeatable read):在一个事务的两次查询之中数据不一致，这可能是两次查询过程中间插入了一个事务更新的原有的数据。 幻读(Phantom Read):在一个事务的两次查询中数据笔数不一致，例如有一个事务查询了几列(Row)数据，而另一个事务却在此时插入了新的几列数据，先前的事务在接下来的查询中，就会发现有几列数据是它先前所没有的。 mysql加锁机制: 根据类型可分为共享锁（SHARED LOCK）和排他锁（EXCLUSIVE LOCK）或者叫读锁（READ LOCK）和写锁（WRITE LOCK）。 根据粒度划分又分表锁和行锁。表锁由数据库服务器实现，行锁由存储引擎实现。 mysql提供了3种事务型存储引擎，InnDB、NDB Cluster和Falcon。 一个事务执行的任何过程中都可以获得锁，但是只有事务提交或回滚的时候才释放这些锁。这些都是隐式锁定，也可以显式锁定，InnoDB支持显式锁定，例如： SELECT …. LOCK IN SHARE MODE (加共享锁) SELECT …..FOR UPDATE(加排他锁) ...

redis stream stream是一个看起来比pubsub可靠多的消息队列。pubsub不靠谱？ 很不靠谱，网络一断或buffer一大就会主动清理数据。stream的设计参考了kafka的消费组模型，redis作者antirez也专门写了篇短文描述了这个过程。 说起新鲜的redis streams，其实Antirez在几年前开了一个新项目叫做disque, 也是用来做消息队列的，奈何没怎么有人关注。我作为antirez的粉丝，肯定是用过了，还tmd改过disque python的库。现在redis5的stream里有一些disque的影子。 更多streams的信息 https://redis.io/topics/streams-intro Redis5.0最近被作者突然放出来了，增加了很多新的特色功能。而Redis5.0最大的新特性就是多出了一个数据结构Stream，它是一个新的强大的支持多播的可持久化的消息队列，作者坦言Redis Stream狠狠地借鉴了Kafka的设计。 Redis Stream 有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的ID和对应的内容。消息是持久化的，Redis重启后，内容还在。 每个Stream都有唯一的名称，它就是Redis的key，在我们首次使用xadd指令追加消息时自动创建。 每个Stream都可以挂多个消费组，每个消费组会有个游标 last_delivered_id 在 Stream 数组之上往前移动，表示当前消费组已经消费到哪条消息了。每个消费组都有一个Stream内唯一的名称，消费组不会自动创建，它需要单独的指令 xgroup create进行创建，需要指定从Stream的某个消息ID开始消费，这个ID用来初始化last_delivered_id变量。 每个消费组(Consumer Group)的状态都是独立的，相互不受影响。也就是说同一份Stream内部的消息会被每个消费组都消费到。 同一个消费组(Consumer Group)可以挂接多个消费者(Consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。每个消费者者有一个组内唯一名称。 消费者(Consumer)内部会有个状态变量pending_ids，它记录了当前已经被客户端读取的消息，但是还没有ack。如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack，它就开始减少。这个pending_ids变量在Redis官方被称之为PEL，也就是Pending Entries List，这是一个很核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理。 http://xiaorui.cc/2018/06/07/%E6%B5%85%E5%85%A5%E6%B5%85%E5%87%BAredis5-0%E7%9A%84streams%E6%95%B0%E6%8D%AE%E7%BB%93%E6%9E%84/ https://toutiao.io/posts/2agvp3/preview

MySQL int 长度 “浮点型"的长度是用来限制数字存储范围的. 比如 float(3,2) 只能够写入 0.00~999.99. “整型"的长度并不会限制存储的数字范围. 比如, int 和 int(3) 的存储范围都是 -2147483648 ~ 2147483647, int unsigned 和 int(3) unsigned 的存储范围都是0 ~ 4294967295. “整型"的长度实际上可以理解为"显示长度”, 如果该字段开启 “Zerofill/补零"就能很明显地知道它的作用. 参考 “高性能MySQL” 的说明 “高性能MySQL” 书中在"4.1 选择优化的数据类型"中提到: MySQL 可以为整数类型指定宽度, 例如 INT(11), 对大多数应用这是没有意义的: 它不会限制值的合法范围, 只是规定了 MySQL 的一些交互工具(例如 MySQL 命令行客户端)用来显示字符的个数. 对于存储和计算来说, INT(1) 和 INT(20) 是相同的 “MySQL 手册"的说明 MySQL 5.7 手册 “12.2.5 Numeric Type Attributes”: MySQL supports an extension for optionally specifying the display width of integer data types in parentheses following the base keyword for the type. For example, INT(4) specifies an INT with a display width of four digits. This optional display width may be used by applications to display integer values having a width less than the width specified for the column by left-padding them with spaces. (That is, this width is present in the metadata returned with result sets. Whether it is used or not is up to the application.) ...

Kafka, offset offset的保存 一个消费组消费partition，需要保存offset记录消费到哪，以前保存在zk中，由于zk的写性能不好，以前的解决方法都是consumer每隔一分钟上报一次。这里zk的性能严重影响了消费的速度，而且很容易出现重复消费。 在0.10版本后，kafka把这个offset的保存，从zk总剥离，保存在一个名叫__consumeroffsets topic的topic中。写进消息的key由groupid、topic、partition组成，value是偏移量offset。topic配置的清理策略是compact。总是保留最新的key，其余删掉。一般情况下，每个key的offset都是缓存在内存中，查询的时候不用遍历partition，如果没有缓存，第一次就会遍历partition建立缓存，然后查询返回。 作者: 123archu 链接: https://www.jianshu.com/p/d3e963ff8b70 来源: 简书 简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。 https://blog.csdn.net/u012129558/article/details/80075270 对Kafka offset的管理，一直没有进行系统的总结，这篇文章对它进行分析。 什么是offset offset是consumer position，Topic的每个Partition都有各自的offset. Keeping track of what has been consumed, is, surprisingly, one of the key performance points of a messaging system. Most messaging systems keep metadata about what messages have been consumed on the broker. That is, as a message is handed out to a consumer, the broker either records that fact locally immediately or it may wait for acknowledgement from the consumer. This is a fairly intuitive choice, and indeed for a single machine server it is not clear where else this state could go. Since the data structure used for storage in many messaging systems scale poorly, this is also a pragmatic choice-since the broker knows what is consumed it can immediately delete it, keeping the data size small. ...

golang kafka https://github.com/Shopify/sarama https://github.com/bsm/sarama-cluster

Kafka 文件存储机制 Kafka 是什么 Kafka 是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统(也可以当做MQ系统)，常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。 1.前言 一个商业化消息队列的性能好坏，其文件存储机制设计是衡量一个消息队列服务技术水平和最关键指标之一。 下面将从Kafka文件存储机制和物理结构角度，分析Kafka是如何实现高效文件存储，及实际应用效果。 2.Kafka文件存储机制 Kafka部分名词解释如下: Broker: 消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。 Topic: 一类消息，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能够同时负责多个topic的分发。 Partition: topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。 Segment: partition物理上由多个segment组成，下面2.2和2.3有详细说明。 offset: 每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息. 分析过程分为以下4个步骤: topic中partition存储分布 partiton中文件存储方式 partiton中segment文件存储结构 在partition中如何通过offset查找message 通过上述4过程详细分析，我们就可以清楚认识到kafka文件存储机制的奥秘。 2.1 topic中partition存储分布 假设实验环境中Kafka集群只有一个broker，xxx/message-folder为数据文件存储根目录，在Kafka broker中server.properties文件配置(参数log.dirs=xxx/message-folder)，例如创建2个topic名称分别为report_push、launch_info, partitions数量都为partitions=4 存储路径和目录规则为: 复制代码 xxx/message-folder |--report_push-0 |--report_push-1 |--report_push-2 |--report_push-3 |--launch_info-0 |--launch_info-1 |--launch_info-2 |--launch_info-3 复制代码 在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1。 如果是多broker分布情况，请参考kafka集群partition分布原理分析 2.2 partiton中文件存储方式 下面示意图形象说明了partition中文件存储方式: image 图1 每个partion(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。 每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。 这样做的好处就是能快速删除无用文件，有效提高磁盘利用率。 2.3 partiton中segment文件存储结构 读者从2.2节了解到Kafka文件系统partition存储方式，本节深入分析partion中segment file组成和物理结构。 segment file组成: 由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀".index"和".log"分别表示为segment索引文件、数据文件. ...

redis config redis 配置 sample, 单机 redis server 配置 bind 0.0.0.0 protected-mode no port 6379 tcp-backlog 511 timeout 0 tcp-keepalive 300 daemonize no pidfile /var/run/redis_6379.pid loglevel notice logfile "" databases 16 always-show-logo no set-proc-title yes proc-title-template "{title} {listen-addr} {server-mode}" stop-writes-on-bgsave-error yes rdbcompression yes rdbchecksum yes dbfilename dump.rdb rdb-del-sync-files no dir /var/lib/redis replica-serve-stale-data yes replica-read-only yes repl-diskless-sync yes repl-diskless-sync-delay 5 repl-diskless-sync-max-replicas 0 repl-diskless-load disabled repl-disable-tcp-nodelay no replica-priority 100 acllog-max-len 128 lazyfree-lazy-eviction no lazyfree-lazy-expire no lazyfree-lazy-server-del no replica-lazy-flush no lazyfree-lazy-user-del no lazyfree-lazy-user-flush no oom-score-adj no oom-score-adj-values 0 200 800 disable-thp yes appendonly no appendfilename "appendonly.aof" appenddirname "appendonlydir" appendfsync everysec no-appendfsync-on-rewrite no auto-aof-rewrite-percentage 100 auto-aof-rewrite-min-size 64mb aof-load-truncated yes aof-use-rdb-preamble yes aof-timestamp-enabled no slowlog-log-slower-than 10000 slowlog-max-len 128 latency-monitor-threshold 0 notify-keyspace-events "" hash-max-listpack-entries 512 hash-max-listpack-value 64 list-max-listpack-size -2 list-compress-depth 0 set-max-intset-entries 512 zset-max-listpack-entries 128 zset-max-listpack-value 64 hll-sparse-max-bytes 3000 stream-node-max-bytes 4096 stream-node-max-entries 100 activerehashing yes client-output-buffer-limit normal 0 0 0 client-output-buffer-limit replica 256mb 64mb 60 client-output-buffer-limit pubsub 32mb 8mb 60 hz 10 dynamic-hz yes aof-rewrite-incremental-fsync yes rdb-save-incremental-fsync yes jemalloc-bg-thread yes https://raw.githubusercontent.com/redis/redis/6.0/redis.conf ...

聚合函数 aggregation function, UNION, UNION ALL 聚合函数 (aggregation function) -也就是组函数 在一个行的集合 (一组行) 上进行操作,对每个组给一个结果。 集合查询操作 union 去重 union用于把两个或者多个select查询的结果集合并成一个 默认情况下,UNION = UNION DISTINCT 进行合并的两个查询,其SELECT列表必须在数量和对应列的数据类型上保持一致; 默认会去掉两个查询结果集中的重复行；默认结果集不排序； 最终结果集的列名来自于第一个查询的SELECT列表 union all 不去重 UNION ALL 不去掉结果集中重复的行 SELECT ... UNION [ALL | DISTINCT] SELECT ... [UNION [ALL | DISTINCT] SELECT ...] 常用的组函数: AVG([distinct] expr) 求平均值 COUNT({*|[distinct] } expr) 统计行的数量 MAX([distinct] expr) 求最大值 MIN([distinct] expr) 求最小值 SUM([distinct] expr) 求累加和 ①每个组函数接收一个参数 ②默认情况下,组函数忽略列值为null的行,不参与计算 ③有时,会使用关键字distinct剔除字段值重复的条数 注意: 当使用组函数的select语句中没有group by子句时,中间结果集中的所有行自动形成一组,然后计算组函数； 组函数不允许嵌套,例如: count(max(…))； 组函数的参数可以是列或是函数表达式； ...

kafka, rabbitmq http://www.infoq.com/cn/articles/kafka-vs-rabbitmq?utm_campaign=infoq_content&utm_source=infoq&utm_medium=feed&utm_term=global

kafka cluster version: 3.0.0 # net podman run \ --name zookeeper \ -p 2181:2181 \ -v /etc/localtime:/etc/localtime:ro \ -v zookeeper-conf:/conf \ -v zookeeper-data:/data \ -v zookeeper-datalog:/datalog \ -e ZOO_4LW_COMMANDS_WHITELIST=* \ -d \ zookeeper:3.7.0 podman run -d --name kafka1 \ -e ALLOW_PLAINTEXT_LISTENER=yes \ -e KAFKA_CFG_ZOOKEEPER_CONNECT=192.168.50.169:2181 \ -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.50.169:9092 \ -p 9092:9092 \ bitnami/kafka:3.2.0 podman run -d --name kafka2 \ -e ALLOW_PLAINTEXT_LISTENER=yes \ -e KAFKA_CFG_ZOOKEEPER_CONNECT=192.168.50.169:2181 \ -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.50.169:9093 \ -p 9093:9092 \ bitnami/kafka:3.2.0 podman run -d --name kafka3 \ -e ALLOW_PLAINTEXT_LISTENER=yes \ -e KAFKA_CFG_ZOOKEEPER_CONNECT=192.168.50.169:2181 \ -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.50.169:9094 \ -p 9094:9092 \ bitnami/kafka:3.2.0 https://github.com/bitnami/bitnami-docker-kafka/blob/master/README.md ...

MySQL using having http://wuzhangshu927.blog.163.com/blog/static/1142246872010113093426574/ USING() using 可用在 join 语句相同字段连接, 起到和 ON 相同作用, inner join 和 left join 中都可以使用 LEFT JOIN 正常写法: SELECT t1.id,t2.name FROM t1 LEFT JOIN t2 ON t1.id=t2.id WHERE .... 其实也可以这么写: SELECT t1.id,t2.name FROM t1 LEFT JOIN t2 USING(id) WHERE .... HAVING MySQL 中的 where 和 having 子句都可以实现筛选记录的功能, having 可以认为是对 where 的补充, 因为它可以对分组数据进行再次判断, 一般跟在 group by 后面, 并可以使用聚集函数 (sum, min, max, avg, count) SELECT `uid`, SUM(`points`) num FROM table GROUP BY `uid` HAVING num > 1000

python grpc grpcio==1.48.2 grpc 会忽略 linux 环境变量 里配置的 no_proxy, 导致请求失败, 在启动客户端之前 临时删除环境变量 unset HTTP_PROXY golang grpc gRPC 通过 HTTP2 协议传输 定义协议 protobuf syntax = "proto3"; message SearchRequest { string query = 1; int32 page_number = 2; int32 results_per_page = 3; } 生成 python 文件 python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. logServer.proto python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. .\tests\remote\grpc_wrapper\service.proto golang grpc install protocol compiler plugins 在 https://pkg.go.dev/google.golang.org/protobuf/cmd/protoc-gen-go 可以看到最新的版本号 go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.27.1 go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.1.0 # 安装后查看 版本 ./protoc-gen-go --version ./protoc-gen-go-grpc --version protoc --go_out=. --go_opt=paths=source_relative \ --go-grpc_out=. --go-grpc_opt=paths=source_relative \ helloworld/helloworld.proto protoc -I proto/ proto/helloworld.proto -go_out=plugins=grpc:proto grpc-dump go install github.com/bradleyjkemp/grpc-tools/grpc-dump@latest grpc-dump --port=12345 http_proxy=http://localhost:12345 my-app wireshark grpc https://mp.weixin.qq.com/s/BdcFRO58ytrtcpYZVT1ymQ ...

influxdb basic influxdb v2 install docker compose yaml # compose.yaml services: influxdb2: image: influxdb:2.7.10-alpine ports: - 8086:8086 environment: DOCKER_INFLUXDB_INIT_MODE: setup DOCKER_INFLUXDB_INIT_USERNAME_FILE: /run/secrets/influxdb2-admin-username DOCKER_INFLUXDB_INIT_PASSWORD_FILE: /run/secrets/influxdb2-admin-password DOCKER_INFLUXDB_INIT_ADMIN_TOKEN_FILE: /run/secrets/influxdb2-admin-token DOCKER_INFLUXDB_INIT_ORG: docs DOCKER_INFLUXDB_INIT_BUCKET: home secrets: - influxdb2-admin-username - influxdb2-admin-password - influxdb2-admin-token volumes: - type: volume source: influxdb2-data target: /var/lib/influxdb2 - type: volume source: influxdb2-config target: /etc/influxdb2 secrets: influxdb2-admin-username: file: ~/.env.influxdb2-admin-username influxdb2-admin-password: file: ~/.env.influxdb2-admin-password influxdb2-admin-token: file: ~/.env.influxdb2-admin-token volumes: influxdb2-data: influxdb2-config: sudo docker compose -f influxdb-compose.yaml up influxdb2 influxdb http api # http api curl -G 'http://localhost:8086/query?pretty=true' --data-urlencode "q=show databases" curl -i -XPOST http://localhost:8086/query --data-urlencode "db=mydb" --data-urlencode "q=CREATE DATABASE db0" curl -i -XPOST "http://192.168.97.1:8086/write?db=monitor" --data-binary 'measurement_0,location=us-midwest temperature=82 1594349970000000000' curl -x http://127.0.0.1:8899/ -i -XPOST "http://192.168.97.1:8086/write?db=monitor" --data-binary 'measurement_0,location=us-midwest temperature=86 1594349970000000000' database management #show db show databases # 创建数据库，同时配置retention policy # DURATION: 数据生命周期 30天 # SHARD DURATION: 分片周期 1小时 # NAME: retention policies 名 CREATE DATABASE "database0" WITH DURATION 30d REPLICATION 1 SHARD DURATION 1h NAME "default" # create db # 数据库名不能包含"-", 可以用"_" create database db0 drop database db0 insert -- float value INSERT measurement0,tag0=tag_value0 field0=1,field1=field_value1 -- int value INSERT measurement0,tag0=tag_value0 field0=1i,field1=field_value1 -- with time INSERT measurement0,tag0=tag_value0 field0=1,field1=field_value1 1570611600000000000 delete DELETE FROM measurement0 WHERE tag0=tag_value0 导出 influx -database 'db0' -execute "select field0,\"field1\" from measurement0 where tag-name0='tag-value0' and time>'2018-05-05 02:00:00' and time<'2018-05-07 11:00:00' order by time tz('Etc/GMT-8')" -format 'csv' -precision 'rfc3339' > xxx.csv # 使用 tz('Etc/GMT-8') 后,过滤条件中的time>'xxx' 填写东8区时间. 安装, influxdb install & config dnf sudo tee /etc/yum.repos.d/influxdb.repo<<EOF [influxdb] name = InfluxDB Repository baseurl = https://repos.influxdata.com/rhel/7/x86_64/stable/ enabled = 1 gpgcheck = 1 gpgkey = https://repos.influxdata.com/influxdb.key EOF dnf install influxdb vim /etc/influxdb/influxdb.conf systemctl enable influxdb # install # archlinux yay -S influxdb # centos # https://www.influxdata.com/blog/package-repository-for-linux/ sudo yum install influxdb sudo yum localinstall influxdb-1.2.4.x86_64.rpm # for Linux Binaries (64-bit) tar xvfz influxdb-1.3.1_linux_amd64.tar.gz rsync -r /path/to/influxdb-1.2.4-1/ / #edit config file emacs /etc/influxdb/influxdb.conf /etc/influxdb/influxdb.conf #reporting-disabled = false [meta] dir = "/var/lib/influxdb/influxdb/meta" #retention-autocreate = true [data] dir = "/var/lib/influxdb/influxdb/data" wal-dir = "/var/lib/influxdb/influxdb/wal" wal-fsync-delay = "100ms" # index-version = "inmem" index-version = "tsi1" trace-logging-enabled = false query-log-enabled = true cache-max-memory-size = "512m" cache-snapshot-memory-size = "32m" # 超过10分钟没有写入, 把cache写到新的TSM文件 cache-snapshot-write-cold-duration = "10m" [coordinator] #慢查询 log-queries-after = "10s" [retention] #edit file /etc/default/influxdb STDERR=/data/logs/influxdb/influxdb.log #edit logrotate config, modify log path /etc/logrotate.d/influxdb # chown chown influxdb:influxdb /data/influxdb/ chown influxdb:influxdb /data/logs/influxdb/ #start systemctl start influxdb #or sudo influxd #connect via cli, rfc3339: 日期格式YYYY-MM-DDTHH:MM:SS.nnnnnnnnnZ influx -precision rfc3339 docker docker pull influxdb:2.7.10-alpine docker run -d \ --name influxdb \ -p 8086:8086 \ -v "influxdb-data:/var/lib/influxdb2" \ -v "influxdb-config:/etc/influxdb2" \ influxdb:2.7.10-alpine # docker docker run -d \ --name influxdb \ --net net0 \ --ip 192.168.1.xxx \ -p 8086:8086 \ -p 8083:8083 \ -v influxdb-config:/etc/influxdb:ro \ -v influxdb-storage:/var/lib/influxdb \ -v /etc/localtime:/etc/localtime:ro \ influxdb # podman podman run -d \ --name influxdb \ -p 8086:8086 \ -p 8083:8083 \ -p 25826:25826 \ -v influxdb-config:/etc/influxdb:ro \ -v influxdb-storage:/var/lib/influxdb \ -v /etc/localtime:/etc/localtime:ro \ influxdb:1.8.10-alpine # in pod podman run -d \ --name influxdb \ --pod monitor \ -v influxdb-config:/etc/influxdb:ro \ -v influxdb-storage:/var/lib/influxdb \ -v /etc/localtime:/etc/localtime:ro \ influxdb chronograf podman run -d \ --name chronograf \ --pod monitor \ -v chronograf:/var/lib/chronograf \ -v /etc/localtime:/etc/localtime:ro \ chronograf --influxdb-url=http://monitor:8086 run influx sudo podman exec -it influxdb influx sudo podman run -it --rm influxdb influx -host influxdb.wiloon.com retention policies show retention policies show retention policies on db0 CREATE RETENTION POLICY "default" ON db0 DURATION 30d REPLICATION 1 SHARD DURATION 1d DEFAULT # ALTER RETENTION POLICY "<policy name>" ON <database> DURATION <duration> REPLICATION 1 SHARD DURATION <shard group duration> DEFAULT ALTER RETENTION POLICY "default" ON db0 DURATION 3h REPLICATION 1 SHARD DURATION 1h DEFAULT # policy name: retention policy 名: default # database: 库名 # duration 3h: 保留3个小时的数据 # shard group duration 1h: 每1个小时的数据一个分片 # DEFAULT: 设置此策略为默认策略 shard list shard id show shards DROP SHARD <shard_id_number> measurement show measurements DROP MEASUREMENT <measurement_name> DROP MEASUREMENT "kernel" select select * from ping where time > now()-1s select average_response_ms from ping where time > now()-1s and url='192.168.53.8' select "database",id,retentionPolicy,seriesCreate,writeReq from "shard" WHERE time>now()-20s AND "database"='database0' AND retentionPolicy='default' AND writeReq>0 select * from "database0"."rentention-policies-0"."measurement0" sELECT mean(m1) * 10 FROM metric0."default".m0 WHERE time >= now() - 10m AND host='host0' GROUP BY time(10s), host influx influx -execute 'select * from "database0"."retention_policies_0"."measurement0" order by time desc limit 1' # show tag keys SHOW TAG KEYS [ON <database_name>] [FROM_clause] [WHERE <tag_key> <operator> ['<tag_value>' | <regular_expression>]] [LIMIT_clause] [OFFSET_clause] show tag keys show tag keys on db0 from measurements0 show tag values on db0 from measure0 with key="host" SHOW FIELD KEYS [ON <database_name>] [FROM <measurement_name>] show tag values from cpu with key=host where service_name=~/xxx/ influx -precision rfc3339 select * from m0 where tag0='tag-value0' and time > '2018-05-16 13:00:00' and time < '2018-05-16 13:01:00' tz('Etc/GMT-8') #use use db0 drop series from series_name select f0,f1 from s0 where t0=~/xxx.*/ ERR: error parsing query: found /, expected regex at line 1, char 56 # =~和后面的正则表达式之间要有空格T_T select f0,f1 from s0 where t0=~ /xxx.*/ 协议, 整数, 浮点数 浮点数 —— 默认是浮点数，InfluxDB假定收到的所有field value都是浮点数。 以浮点类型存储上面的82： ...

时间序列数据库 存储大量时间相关的数据(如日志，用户行为等) 创造了一种新型的数据库分类——时间序列数据库(Time Series Database). 时间序列数据库主要用于指处理带时间标签 (按照时间的顺序变化，即时间序列化）的数据，带时间标签的数据也称为时间序列数据。 influxdb influxdb是最新的一个时间序列数据库,最新一两年才产生,但已经拥有极高的人气。influxdb 是用Go写的,现在v0.9正在开发中,之前开源出来的最稳定的版本是0.88的,但是0.8X是没有集群方案的,但在0.9中会加入进来。 0.9版本的influxdb对于之前会有很大的改变,后端存储有LevelDB换成了BoltDB,读写的API也是有了很大的变化,也将支持集群化,continuous query,支持retention policy,读写性能也是哇哇的,可以说是时间序列存储的完美方案,但是由于还很年轻,可能还会存在诸多的问题,就像现在正在开发的0.9一样,发布一拖再拖,就是由于还有些技术壁垒没有攻陷。 对于influxdb我不想多说些什么,之后打算开一个专题,专门详细来说一说这个玩意,因为我看国内几乎没有详细的文章来讲influxdb的。 http://www.opscoder.info/tsdb.html 对于时间序列的存储,一般会采用专门的时间序列数据库,而不会去使用MySQL或是mongo(但zabbix就是用的MySQL,所以它在IO上面遇到了瓶颈)。现在时间序列的数据库是有很多的,比如graphite、opentsdb以及新生的influxdb。最近也相继研究了一下这三个数据库,现在把研究所得记录下来。 特性： 高效的时间序列数据写入性能。自定义TSM引擎，快速数据写入和高效数据压缩。 无额外存储依赖。 简单，高性能的HTTP查询和写入API。 以插件方式支持许多不同协议的数据摄入，如：graphite，collectd，和openTSDB SQL-like查询语言，简化查询和聚合操作。 索引Tags，支持快速有效的查询时间序列。 保留策略有效去除过期数据。 连续查询自动计算聚合数据，使频繁查询更有效。 influxDB 不支持数据库的更新操作，毕竟时间数据只能随着时间产生新数据，肯定无法对过去的数据修改。 适合于写多读少的场景 graphite graphite算是一个老牌的时间序列存储解决方案了,graphite由三个部分组成,分别是carbon、whisper和graphite web carbon:实际上是一系列守护进程,这些守护进程用Twisted的事件驱动网络引擎监听时间序列数据。Twisted框架让Carbon守护进程能够以很低的开销处理大量的客户端和流量。 whisper:是一个用于存储时间序列数据的数据库,之后应用程序可以用create,update和fetch操作获取并操作这些数据。 graphite web:使用django开发的一套web,提供一些常用的聚合函数,可以界面友好的展示出图形。再盗图: whisper支持RRD,可以很方便的定义retention,以及定义storage scheme,不需要手动做,graphite会自动帮你按照不同的scheme实现aggregation。这样每个metric的大小就是固定的了,所以理论上可以永久存储数据。graphite的集群方案主要有两种,分别是使用graphite自带的relay或是使用第三方的工具。 当使用自带的relay时,只需要在配置文件中配置要relay到哪些机器即可,这样在数据写入的时候,被写入的节点会relay一份到这些机器中。在读取的时候,在graphite web的settings中配置HOSTS的列表,这样在django的web中会依次从这些HOSTS中读出数据。 另一种是需用第三方的relay工具,Booking公司开源出来了他们所用的用C写得 carbon-c-relay,以及用GO写得carbon-relay-ng。其基本的思想是运用一致性哈希,可以将不同的metric发到不同的机器,以来达到集群的目的,据Booking称他们的graphite集群的规模达到了百台机器,而mertic也达到了百万的级别. 以上的两种方式都会遇到了一个共同的问题,就是集群扩容的问题,我不知道Booking是怎么来解决这个问题的,但是我这边目前是想到了几种方式,这个在我前一篇的文章中已经阐述了: graphite集群扩容方案探究,在这里就先不赘述了。 最后说一下,除了集群问题以外,graphite的还有一性能问题就是读的性能稍差,这决定于其存储的方式,其实在读的时候会去读whisper文件(虽然在django层做了缓存,但是缓存的功能比较弱),通过seek的方式来获取数据的位置,在将数据取出。 opentsdb Opentsdb是一个基于Hbase的时间序列数据库 (新版也支持Cassandra）。 其基于Hbase的分布式列存储特性实现了数据高可用，高性能写的特性。受限于Hbase，存储空间较大，压缩不足。依赖整套 HBase, ZooKeeper 采用无模式的tagset数据结构(sys.cpu.user 1436333416 23 host=web01 user=10001) 结构简单，多value查询不友好 HTTP-DSL查询 opentsdb是一个比较重的时间序列解决方案,为什么说他重呢？因为它的组成是这样的: 可以看到opentsdb所依赖的存储是Hbase集群。TSD在其中担任的责任是IO部分,TSD其实就是一个后台的daemon,一般我会会用一组TSD达成一个TSD的集群 (其实不能算是集群) ,没有master/slave之分,也没有共享状态,然后在之前用LB设备来做负载均衡,官方比较推荐的是用varnish。 当让后端的存储也可以用其他的例如hadoop,但是官方还是建议用Hbase,因为opentsdb就是Hbase社区孵化出来的产品。那么问题来了,Hbase的运维将是一个艰巨的任务,这个依赖于 zookeeper 搭建的集群的坑还是很多的,我看了一下官方文档就有上千页。这里面的优化维护需要有专业的Hbase专家才能完成。 另外opentsdb做不到graphite那样自动做downsample,也就是做不到RRD那样地去存储数据,需要在外面自己做一层手动干这活,再把聚合后的数据写入Hbase,唉,还真是一个费时费力的活。 除此之外opentsdb还是很不错的,读写性能挺高,而且支持tag,支持ttl,支持各种聚合函数。现在很多的监控的metric的存储都是用的opentsdb,嗯,是的,只要有能力玩转还是个不错的选择。 Timescale 基于传统关系型数据库postgresql改造的时间序列数据库 一款兼容sql的时序数据库， 底层存储架构在postgresql上。 作为一个postgresql的扩展提供服务。 ...

InfluxDB 时间序列数据库 InfluxDB 基于 Go 语言开发,社区非常活跃,项目更新速度很快,日新月异,关注度高。 特点 可以设置metric的保存时间。 支持通过条件过滤以及正则表达式删除数据。 支持类似 sql 的语法。 可以设置数据在集群中的副本数。 支持定期采样数据,写入另外的measurement,方便分粒度存储数据。 概念 数据格式 Line Protocol measurement[,tag_key1=tag_value1…] field_key=field_value[,field_key2=field_value2] [timestamp] cpu_load,host_id=1 value=0.1 1434055562000000000 相比于 JSON 格式,无需序列化,更加高效。 measurement: metric name,例如 cpu_load。 field-key, field-value: 通常用来存储数据,类似 opentsdb 中的 value=0.6,但是支持各种类型,数据存储时不会进行索引,每条数据必须拥有一个 field-key,如果使用 field-key 进行过滤,需要遍历一遍所有数据。 tags-key, tags-value: 和 field-key 类似,但是会进行索引,方便查询时用于过滤条件。 Series measurement, tag set, retention policy 相同的数据集合算做一个 series。 假设 cpu_load 有两个 tags,host_id 和 name,host_id 的数量为 100,name 的数量为 200,则 series 基数为 100 * 200 = 20000。 ...

MVCC, Multiversion Concurrency Control, 多版本并发控制 http://donghui.blog.51cto.com/2709336/692586 多版本并发控制技术已经被广泛运用于各大数据库系统中,如 Oracle, MS SQL Server 2005+, Postgresql, Firebird, Maria 等等, 开源数据库 MySQL 中流行的 INNODB 引擎也采用了类似的并发控制技术.本文就将结合实例来解析不同事务隔离等级下 INNODB 的 MVCC 实现原理. 1.1 MVCC 简介 MVCC (Multiversion Concurrency Control), 即多版本并发控制技术, 它使得大部分支持行锁的事务引擎, 不再单纯的使用行锁来进行数据库的并发控制, 取而代之的是, 把数据库的行锁与行的多个版本结合起来, 只需要很小的开销, 就可以实现非锁定读, 从而大大提高数据库系统的并发性能. 1.2 实现原理 MVCC 可以提供基于某个时间点的快照, 使得对于事务看来, 总是可以提供与事务开始时刻相一致的数据, 而不管这个事务执行的时间有多长. 所以在不同的事务看来, 同一时刻看到的相同行的数据可能是不一样的, 即一个行可能有多个版本. 是否听起来不可思议呢? 原来, 为了实现 mvcc, innodb 对每一行都加上了两个隐含的列, 其中一列存储行被更新的"时间", 另外一列存储行被删除的"时间". 但是 innodb 存储的并不是绝对的时间, 而是与时间对应的数据库系统的版本号, 每当一个事务开始的时候, innodb 都会给这个事务分配一个递增的版本号, 所以版本号也可以被认为是事务号. 对于每一个"查询"语句, innodb 都会把这个查询语句的版本号同这个查询语句遇到的行的版本号进行对比, 然后结合不同的事务隔离等级, 来决定是否返回该行. ...

kafka config, server, producer 大体上来说,用户首先构建待发送的消息对象ProducerRecord,然后调用KafkaProducer#send方法进行发送。KafkaProducer接收到消息后首先对其进行序列化,然后结合本地缓存的元数据信息一起发送给partitioner去确定目标分区,最后追加写入到内存中的消息缓冲池(accumulator)。此时KafkaProducer#send方法成功返回。 KafkaProducer中还有一个专门的Sender IO线程负责将缓冲池中的消息分批次发送给对应的broker,完成真正的消息发送逻辑。 基本设计特点 结合源代码,笔者认为新版本的producer从设计上来说具有以下几个特点(或者说是优势): 总共创建两个线程: 执行KafkaPrducer#send逻辑的线程——我们称之为"用户主线程"；执行发送逻辑的IO线程——我们称之为"Sender线程" 不同于Scala老版本的producer,新版本producer完全异步发送消息,并提供了回调机制(callback)供用户判断消息是否成功发送 batching机制——“分批发送"机制。每个批次(batch)中包含了若干个PRODUCE请求,因此具有更高的吞吐量 更加合理的默认分区策略: 对于无key消息而言,Scala版本分区策略是一段时间内(默认是10分钟)将消息发往固定的目标分区,这容易造成消息分布的不均匀,而新版本的producer采用轮询的方式均匀地将消息分发到不同的分区 底层统一使用基于 Selector 的网络客户端实现, 结合 Java 提供的 Future 实现完整地提供了更加健壮和优雅的生命周期管理。 其实,新版本 producer 的设计优势还有很多, 诸如监控指标更加完善等这样的就不一一细说了。总之, 新版本producer更加地健壮,性能更好~ Kafka新版本clients在设计底层网络库时采用了Java的Selector机制，而后者在 Linux上的实现机制就是 epoll；但是在 Windows平台上，Java NIO的 Selector 底层是使用 select模型而非IOCP 实现的，只有 Java NIO2才是使用 IOCP 实现的。因此在这一点上，在 Linux 上部署Kafka要比在Windows上部署能够得到更高效的I/O处理性能。 properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer<>(props); for(int i = 0; i < 100; i++) producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i))); producer.close(); bootstrap.servers 配置kafka 查询集群 metadata 服务的地址, 建立连接时,kafa producer向 bootstrap.servers 发 metadata 请求, 从 返回 的metadata response里得到kafka 集群的地址,再建立连接 ...