腾讯云kafka（腾讯云游戏start）

本文目录一览：

• 1、东莞腾讯云代理商，腾讯云消息队列Kafka怎么样呢？
• 2、Elasticsearch数据迁移与集群容灾
• 3、东莞腾讯云代理商，腾讯云消息队列 Kafka怎么样呢？
• 4、腾讯云CKafka压测踩坑记录
• 5、腾讯云Logstash实战4-使用Logstash消费kafka数据并写入到Elasticsearch

东莞腾讯云代理商，腾讯云消息队列Kafka怎么样呢？

腾讯云Kafka分专业版和标准版，支持多种场景使用，腾讯云Kafka，还不错，主要是不贵，也是咨询的腾讯云时鱼科技开的。。你的采纳 是我回答的动力。

Elasticsearch数据迁移与集群容灾

本文讨论如何跨集群迁移ES数据以及如何实现ES的同城跨机房容灾和异地容灾。

在ES的生产实践中，往往会遇到以下问题：

根据业务需求，存在以下场景：

如果是第一种场景，数据迁移过程中可以停止写入，可以采用诸如elasticsearch-dump、logstash、reindex、snapshot等方式进行数据迁移。实际上这几种工具大体上可以分为两类：

如果是第二种场景，数据迁移过程中旧集群不能停止写入，需要根据实际的业务场景解决数据一致性的问题：

下面介绍一下在旧集群可以停止写入的情况下进行数据迁移的几种工具的用法。

elasticsearch-dump是一款开源的ES数据迁移工具，github地址:

以下操作通过elasticdump命令将集群x.x.x.1中的companydatabase索引迁移至集群x.x.x.2。注意第一条命令先将索引的settings先迁移，如果直接迁移mapping或者data将失去原有集群中索引的配置信息如分片数量和副本数量等，当然也可以直接在目标集群中将索引创建完毕后再同步mapping与data

logstash支持从一个ES集群中读取数据然后写入到另一个ES集群，因此可以使用logstash进行数据迁移，具体的配置文件如下：

上述配置文件将源ES集群的所有索引同步到目标集群中，当然可以设置只同步指定的索引，logstash的更多功能可查阅logstash官方文档 logstash 官方文档 .

reindex是Elasticsearch提供的一个api接口，可以把数据从一个集群迁移到另外一个集群。

snapshot api是Elasticsearch用于对数据进行备份和恢复的一组api接口，可以通过snapshot api进行跨集群的数据迁移，原理就是从源ES集群创建数据快照，然后在目标ES集群中进行恢复。需要注意ES的版本问题：

如果旧集群不能停止写入，此时进行在线数据迁移，需要保证新旧集群的数据一致性。目前看来，除了官方提供的CCR功能，没有成熟的可以严格保证数据一致性的在线数据迁移方法。此时可以从业务场景出发，根据业务写入数据的特点选择合适的数据迁移方案。

一般来说，业务写入数据的特点有以下几种：

下面来具体分析不同的写入数据的特点下，该如何选择合适的数据迁移方式。

在日志或者APM的场景中，数据都是时序数据，一般索引也都是按天创建的，当天的数据只会写入当前的索引中。此时，可以先把存量的不再写入的索引数据一次性同步到新集群中，然后使用logstash或者其它工具增量同步当天的索引，待数据追平后，把业务对ES的访问切换到新集群中。

具体的实现方案为：

add only的数据写入方式，可以按照数据写入的顺序(根据_doc进行排序，如果有时间戳字段也可以根据时间戳排序)批量从旧集群中拉取数据，然后再批量写入新集群中；可以通过写程序，使用用scroll api 或者search_after参数批量拉取增量数据，再使用bulk api批量写入。

使用scroll拉取增量数据：

上述操作可以每分钟执行一次，拉起前一分钟新产生的数据，所以数据在旧集群和新集群的同步延迟为一分钟。

使用search_after批量拉取增量数据：

上述操作可以根据需要自定义事件间隔执行，每次执行时修改search_after参数的值，获取指定值之后的多条数据；search_after实际上相当于一个游标，每执行一次向前推进，从而获取到最新的数据。

使用scroll和search_after的区别是：

另外，如果不想通过写程序迁移旧集群的增量数据到新集群的话，可以使用logstash结合scroll进行增量数据的迁移，可参考的配置文件如下：

使用过程中可以根据实际业务的需求调整定时任务参数schedule以及scroll相关的参数。

业务场景如果是写入ES时既有追加，又有存量数据的更新，此时比较重要的是怎么解决update操作的数据同步问题。对于新增的数据，可以采用上述介绍的增量迁移热索引的方式同步到新集群中。对于更新的数据，此时如果索引有类似于updateTime的字段用于标记数据更新的时间，则可以通过写程序或者logstash，使用scroll api根据updateTime字段批量拉取更新的增量数据，然后再写入到新的集群中。

可参考的logstash配置文件如下：

实际应用各种，同步新增(add)的数据和更新(update)的数据可以同时进行。但是如果索引中没有类似updateTime之类的字段可以标识出哪些数据是更新过的，目前看来并没有较好的同步方式，可以采用CCR来保证旧集群和新集群的数据一致性。

如果业务写入ES时既有新增(add)数据,又有更新(update)和删除(delete)数据，可以采用6.5之后商业版X-pack插件中的CCR功能进行数据迁移。但是使用CCR有一些限制，必须要注意：

具体的使用方式如下：

如果业务是通过中间件如kafka把数据写入到ES， 则可以使用如下图中的方式，使用logstash消费kafka的数据到新集群中，在旧集群和新集群数据完全追平之后，可以切换到新集群进行业务的查询，之后再对旧的集群下线处理。

使用中间件进行同步双写的优点是：

当然，双写也可以使用其他的方式解决，比如自建proxy，业务写入时向proxy写入，proxy把请求转发到一个或者多个集群中，但是这种方式存在以下问题：

随着业务规模的增长，业务侧对使用的ES集群的数据可靠性、集群稳定性等方面的要求越来越高，所以要比较好的集群容灾方案支持业务侧的需求。

如果是公司在自建IDC机房内，通过物理机自己搭建的ES集群，在解决跨机房容灾的时候，往往会在两个机房 部署两个ES集群，一主一备，然后解决解决数据同步的问题；数据同步一般有两种方式，一种方式双写，由业务侧实现双写保证数据一致性，但是双写对业务侧是一个挑战，需要保证数据在两个集群都写成功才能算成功。另外一种方式是异步复制，业务侧只写主集群，后台再把数据同步到备集群中去，但是比较难以保证数据一致性。第三种方式是通过专线打通两个机房，实现跨机房部署，但是成本较高。

因为数据同步的复杂性，云厂商在实现ES集群跨机房容灾的时候，往往都是通过只部署一个集群解决，利用ES自身的能力同步数据。国外某云厂商实现跨机房部署ES集群的特点1是不强制使用专用主节点，如上图中的一个集群，只有两个节点，既作为数据节点也作为候选主节点；主分片和副本分片分布在两个可用区中，因为有副本分片的存在，可用区1挂掉之后集群仍然可用，但是如果两个可用区之间网络中断时，会出现脑裂的问题。如下图中使用三个专用主节点，就不会存在脑裂的问题了。

但是如果一个地域没有三个可用区怎么办呢，那就只能在其中一个可用区中放置两个专用主节点了，如国内某云厂商的解决方案：

但是重建节点的过程还是存在问题的，如上图中，集群本身的quorum应该为2，可用区1挂掉后，集群中只剩一个专用主节点，需要把quorum参数(discovery.zen.minimum_master_nodes)调整为1后集群才能够正常进行选主，等挂掉的两个专用主节点恢复之后，需要再把quorum参数(discovery.zen.minimum_master_nodes)调整为2，以避免脑裂的发生。

当然还是有可以把无法选主和脑裂这两个可能发生的问题规避掉的解决方案，如下图中国内某云厂商的解决思路：

创建双可用区集群时，必须选择3个或者5个专用主节点，后台会在一个隐藏的可用区中只部署专用主节点；方案的优点1是如果一个可用区挂掉，集群仍然能够正常选主，避免了因为不满足quorum法定票数而无法选主的情况；2是因为必须要选择三个或5个专用主节点，也避免了脑裂。

想比较一主一备两个集群进行跨机房容灾的方式，云厂商通过跨机房部署集群把原本比较复杂的主备数据同步问题解决了，但是，比较让人担心的是，机房或者可用区之间的网络延迟是否会造成集群性能下降。这里针对腾讯云的双可用区集群，使用标准的benchmark工具对两个同规格的单可用区和双可用区集群进行了压测，压测结果如下图所示：

从压测结果的查询延时和写入延时指标来看，两种类型的集群并没有明显的差异，这主要得益与云上底层网络基础设施的完善，可用区之间的网络延迟很低。

类似于同城跨机房容灾，异地容灾一般的解决思路是在异地两个机房部署一主一备两个集群。业务写入时只写主集群，再异步地把数据同步到备集群中，但是实现起来会比较复杂，因为要解决主备集群数据一致性的问题，并且跨地域的话，网络延迟会比较高；还有就是，当主集群挂掉之后，这时候切换到备集群，可能两边数据还没有追平，出现不一致，导致业务受损。当然，可以借助于kafka等中间件实现双写，但是数据链路增加了，写入延迟也增加了，并且kafka出现问题，故障可能就是灾难性的了。

一种比较常见的异步复制方法是，使用snapshot备份功能，定期比如每个小时在主集群中执行一次备份，然后在备集群中进行恢复，但是主备集群会有一个小时的数据延迟。以腾讯云为例，腾讯云的ES集群支持把数据备份到对象存储COS中，因为可以用来实现主备集群的数据同步，具体的操作步骤可以参考 。

在6.5版本官方推出了CCR功能之后，集群间数据同步的难题就迎刃而解了。可以利用CCR来实现ES集群的异地容灾：

CCR是类似于数据订阅的方式，主集群为Leader, 备集群为Follower, 备集群以pull的方式从主集群拉取数据和写请求；在定义好Follwer Index时，Follwer Index会进行初始化，从Leader中以snapshot的方式把底层的segment文件全量同步过来，初始化完成之后，再拉取写请求，拉取完写请求后，Follwer侧进行重放，完成数据的同步。CCR的优点当然是因为可以同步UPDATE/DELETE操作，数据一致性问题解决了，同步延时也减小了。

另外，基于CCR可以和前面提到的跨机房容灾的集群结合，实现两地多中心的ES集群。在上海地域，部署有多可用区集群实现跨机房的高可用，同时在北京地域部署备集群作为Follwer利用CCR同步数据，从而在集群可用性上又向前走了一步，既实现了同城跨机房容灾，又实现了跨地域容灾。

但是在出现故障时需要把集群的访问从上海切换到北京时，会有一些限制，因为CCR中的Follwer Index是只读的，不能写入，需要切换为正常的索引才能进行写入，过程也是不可逆的。不过在业务侧进行规避，比如写入时使用新的正常的索引，业务使用别名进行查询，当上海地域恢复时，再反向的把数据同步回去。

现在问题就是保证上海地域集群数据的完整性，在上海地域恢复后，可以在上海地域新建一个Follower Index，以北京地域正在进行写的索引为Leader同步数据，待数据完全追平后，再切换到上海地域进行读写，注意切换到需要新建Leader索引写入数据。

数据同步过程如下所示：

1.上海主集群正常提供服务，北京备集群从主集群Follow数据

2.上海主集群故障，业务切换到北京备集群进行读写，上海主集群恢复后从北京集群Follow数据

东莞腾讯云代理商，腾讯云消息队列 Kafka怎么样呢？

腾讯云专业版的消息队列Kafka版实例支持两种Topic存储引擎，云存储和Local存储，腾讯云Kafka，还不错，主要是不贵，也是咨询的时鱼科技开的。

腾讯云CKafka压测踩坑记录

由于最近项目要上腾讯云，不得不对腾讯云CKafka进行压测，评估kafka的处理性能是否满足项目需求。（项目期望Kafka能够处理上千万级别的MQ）

一、 明确测试目的

本次性能测试在UAT环境下腾讯云服务器上CKafka处理MQ消息能力进行压力测试。测试包括对Kafka写入MQ消息和消费MQ消息进行压力测试，根据100w、500w和1000w级别的消息处理结果，评估Kafka的处理性能极限值。

二、 Kafka测试前期准备

2.1  Kafka的性能测试主要测试kafka的吞吐量，kafka吞性能为生产者在向kafka传入消息时的写入量，kafka的吐性能为消费者在kafka集群中消费的能力，也就是读取量。

2.2 Borker相关

        Kafka的borker是kafka集群的缓存代理，消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。具体可参考kafka官方文档。

2.3 Cousumer相关

       Consumer为kafka的消费者，同一个topic消费者越多越快，但是需要注意的是，消费者的数量不能超过topic的分区数量，因为每个topic的每个分区只能被一个消费者消费，多出来的消费者会无信息可消费。导致资源浪费。具体可参考腾讯云Ckafka指南。

三、Kafka常用参数配置

3.1 生产端常用参数配置如下：

消费者参数配置如下：

Broker 配置参数说明如下：

四、场景设计

4.1 Kafka写入消息压力测试

4.2 Kafka消费压测测试

五、测试方法

5.1 在服务器上使用Kafka自带的测试脚本，分别模拟100w、500w和1000w的消息写入请求，查看Kafka处理不同数量级的消息数时的处理能力，包括每秒生成消息数、吞吐量、消息延迟时间。Kafka消息吸入创建的topic命名为test-2，使用命令发起消费该topic的请求，查看Kafka消费不同数量级别的消息时的处理能力。

5.2 压测命令（脚本执行目录：bin/）

            zookeeper脚本：./kafka-consumer-perf-test.sh --zookeeper IP:port --topic forbid_resources_topic --fetch-size 1048576 --messages 10000 --threads 1

 写入脚本命令的参数解析：

消费脚本参数解析：

六、测试结果

写入100W结果：

消费100W结果：

注意：这里的坑就来了实际消费数量与脚本设置的消费数量不一致，在这里的这个问题查了很多资料发现两个问题,一会下面慢慢解释，先来看每个字段的意思。

我们先来看看消费的每个数据字段的含义，如下图：

上图我们可以看出，data.consumed.in.nMsg（总消费消息数）与脚本中messages设置的值不一致；设置消费100W，实际消费121431条消息。

坑就在这里，由于是买的腾讯云的PASS服务，很多东西都没办法获取权限查看，只能一步步和客服沟通，挨个排查。

腾讯售后客服也给发了案例和教程，发现教程里也是实际消费和设置消费数不一致；，此处省略一万字。。。

最终在无意之间更改了Topic分区数之后再次运行脚本发现问题消失了，测试环境的Topic分区设置为1，后续增加分区数发现能实际消费和设置消费消息数一致。最后经过多次测试最终Topic分区数设置为3。这次之后发现忽略了腾讯云提供的压测指南中的底部有几句话；

在多次和腾讯售后客服沟通和交流，后续也有和腾讯相关后端开发沟通，发现竟然连开发都解释不清楚出现这个问题的原因，只要深入了解，就会含糊解释说：“我们卖了这么多产品，Kafka肯定不会有问题的”，其内部也没有关于Kafka相关的压测分析案例。可能是我寡闻，在此记录，也是分享出此次压测的踩坑经历。

腾讯云Logstash实战4-使用Logstash消费kafka数据并写入到Elasticsearch

Logstash的一个典型应用场景，就是消费kafka中的数据并且写入到Elasticsearch, 使用 腾讯云Logstash 产品，可以通过简单的配置快速地完成这一过程。

在“管道管理”页面，点击“新建管道”按钮，创建一个管道：

进入管道配置页面，点击“引用模板”按钮，同时引用“input-kafka”和“output-elasticsearch”两个模板：

在管道配置中，分别针对“input-kafka”和“output-elasticsearch”进行配置，一些关键的配置参数说明如下：

查看更多参数，可以参考 input-kafka

查看更多参数，可以参考 output-elasticsearch

在配置完管道后，点击“保存并部署”创建一个管道并自动部署:

在控制台查看Logstash的运行日志，如果没有ERROR级别的日志，则说明管道运行正常：

进入到output-elasticsearch中定义的输出端的ES集群对应的kibana页面，在Dev tools工具栏里查看索引是否存在，以及索引的文档数量是否正确：