Strimzi Kafka Bridge(桥接)实战之二：生产和发送消息

缠丝猫  金牌会员 | 2023-10-8 07:11:03 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 915|帖子 915|积分 2745

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos
本篇概览


  • 本文是《Strimzi Kafka Bridge(桥接)实战之》系列的第二篇，咱们直奔bridge的重点：常用接口，用实际操作体验如何用bridge完成常用的消息收发业务
  • 官方的openapi接口文档地址 : https://strimzi.io/docs/bridge/in-development/#_openapi
  • 整篇文章由以下内容构成：

  • 准备工作：创建topic
  • 生产消息
  • 消费消息，strimzi bridge消费消息的逻辑略有些特殊，就是要提前创建strimzi bridge consumer，再通过consumer来调用拉取消息的接口


  • 完成本篇实战后，相信您已经可以数量的通过http来使用kafka的服务了
准备工作：创建topic


  • 遗憾的是，bridge未提供创建topic的API，所以咱们还是用命令来创建吧
  • ssh登录kubernetes的宿主机
  • 执行创建名为bridge-quickstart-topic的topic，共四个分区
  1. kubectl -n aabbcc \
  2. run kafka-producer \
  3. -ti \
  4. --image=quay.io/strimzi/kafka:0.32.0-kafka-3.3.1 \
  5. --rm=true \
  6. --restart=Never \
  7. -- bin/kafka-topics.sh \
  8. --bootstrap-server my-cluster-kafka-bootstrap:9092 \
  9. --create \
  10. --topic bridge-quickstart-topic \
  11. --partitions 4 \
  12. --replication-factor 1
复制代码

  • 检查topic创建是否成功
  1. kubectl -n aabbcc \
  2. run kafka-producer \
  3. -ti \
  4. --image=quay.io/strimzi/kafka:0.32.0-kafka-3.3.1 \
  5. --rm=true \
  6. --restart=Never \
  7. -- bin/kafka-topics.sh \
  8. --bootstrap-server my-cluster-kafka-bootstrap:9092 \
  9. --describe \
  10. --topic bridge-quickstart-topic
复制代码

  • 如下图，可见topic的创建符合预期

  • 接下来的操作都是向bridge发送http请求完成的，我这边宿主机的IP地址是192.168.0.1，bridge的NodePort端口号31331
查看指定topic的详情


  • 如下请求，可以取得topicbridge-quickstart-topic的详情
  1. curl -X GET \
  2.   http://192.168.0.1:31331/topics/bridge-quickstart-topic
复制代码

  • 收到响应如下，是这个topic的详细信息
  1. {
  2.         "name": "bridge-quickstart-topic",
  3.         "configs": {
  4.                 "compression.type": "producer",
  5.                 "leader.replication.throttled.replicas": "",
  6.                 "message.downconversion.enable": "true",
  7.                 "min.insync.replicas": "1",
  8.                 "segment.jitter.ms": "0",
  9.                 "cleanup.policy": "delete",
  10.                 "flush.ms": "9223372036854775807",
  11.                 "follower.replication.throttled.replicas": "",
  12.                 "segment.bytes": "1073741824",
  13.                 "retention.ms": "604800000",
  14.                 "flush.messages": "9223372036854775807",
  15.                 "message.format.version": "3.0-IV1",
  16.                 "max.compaction.lag.ms": "9223372036854775807",
  17.                 "file.delete.delay.ms": "60000",
  18.                 "max.message.bytes": "1048588",
  19.                 "min.compaction.lag.ms": "0",
  20.                 "message.timestamp.type": "CreateTime",
  21.                 "preallocate": "false",
  22.                 "min.cleanable.dirty.ratio": "0.5",
  23.                 "index.interval.bytes": "4096",
  24.                 "unclean.leader.election.enable": "false",
  25.                 "retention.bytes": "-1",
  26.                 "delete.retention.ms": "86400000",
  27.                 "segment.ms": "604800000",
  28.                 "message.timestamp.difference.max.ms": "9223372036854775807",
  29.                 "segment.index.bytes": "10485760"
  30.         },
  31.         "partitions": [
  32.                 {
  33.                         "partition": 0,
  34.                         "leader": 0,
  35.                         "replicas": [
  36.                                 {
  37.                                         "broker": 0,
  38.                                         "leader": true,
  39.                                         "in_sync": true
  40.                                 }
  41.                         ]
  42.                 },
  43.                 {
  44.                         "partition": 1,
  45.                         "leader": 0,
  46.                         "replicas": [
  47.                                 {
  48.                                         "broker": 0,
  49.                                         "leader": true,
  50.                                         "in_sync": true
  51.                                 }
  52.                         ]
  53.                 },
  54.                 {
  55.                         "partition": 2,
  56.                         "leader": 0,
  57.                         "replicas": [
  58.                                 {
  59.                                         "broker": 0,
  60.                                         "leader": true,
  61.                                         "in_sync": true
  62.                                 }
  63.                         ]
  64.                 },
  65.                 {
  66.                         "partition": 3,
  67.                         "leader": 0,
  68.                         "replicas": [
  69.                                 {
  70.                                         "broker": 0,
  71.                                         "leader": true,
  72.                                         "in_sync": true
  73.                                 }
  74.                         ]
  75.                 }
  76.         ]
  77. }
复制代码
批量生产消息（同步）


  • 试试bridge提供的批量生产消息的API，以下命令会生产了三条消息，第一条通过key的hash值确定分区，第二条用partition参数明确指定了分区是2，第三条的分区是按照轮询策略更新的
  1. curl -X POST \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic \
  3.   -H 'content-type: application/vnd.kafka.json.v2+json' \
  4.   -d '{
  5.     "records": [
  6.         {
  7.             "key": "my-key",
  8.             "value": "sales-lead-0001"
  9.         },
  10.         {
  11.             "value": "sales-lead-0002",
  12.             "partition": 2
  13.         },
  14.         {
  15.             "value": "sales-lead-0003"
  16.         }
  17.     ]
  18. }'
复制代码

  • bridge响应如下，会返回每一条消息的partition和offset，这就是同步消息的特点，等到meta信息更新完毕后才会返回
  1. {
  2.         "offsets": [{
  3.                 "partition": 0,
  4.                 "offset": 0
  5.         }, {
  6.                 "partition": 2,
  7.                 "offset": 0
  8.         }, {
  9.                 "partition": 3,
  10.                 "offset": 0
  11.         }]
  12. }
复制代码
批量生产消息（异步）


  • 有的场景下，例如追求高QPS并且对返回的meta信息不关注，可以考虑异步的方式发送消息，也就是说bridge收到响应后立即返回200，这种异步模式和前面的同步模式只有一个参数的差别：在请求url中增加async=true即可
  1. curl -X POST \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic?async=true \
  3.   -H 'content-type: application/vnd.kafka.json.v2+json' \
  4.   -d '{
  5.     "records": [
  6.         {
  7.             "key": "my-key",
  8.             "value": "sales-lead-0001"
  9.         },
  10.         {
  11.             "value": "sales-lead-0002",
  12.             "partition": 2
  13.         },
  14.         {
  15.             "value": "sales-lead-0003"
  16.         }
  17.     ]
  18. }'
复制代码

  • 没有响应body，请您自行请求感受一下，响应明显比同步模式快
查看partition


  • 查看tipic的parition情况
  1. curl -X GET \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic/partitions
复制代码

  • 响应
  1. [{
  2.         "partition": 0,
  3.         "leader": 0,
  4.         "replicas": [{
  5.                 "broker": 0,
  6.                 "leader": true,
  7.                 "in_sync": true
  8.         }]
  9. }, {
  10.         "partition": 1,
  11.         "leader": 0,
  12.         "replicas": [{
  13.                 "broker": 0,
  14.                 "leader": true,
  15.                 "in_sync": true
  16.         }]
  17. }, {
  18.         "partition": 2,
  19.         "leader": 0,
  20.         "replicas": [{
  21.                 "broker": 0,
  22.                 "leader": true,
  23.                 "in_sync": true
  24.         }]
  25. }, {
  26.         "partition": 3,
  27.         "leader": 0,
  28.         "replicas": [{
  29.                 "broker": 0,
  30.                 "leader": true,
  31.                 "in_sync": true
  32.         }]
  33. }]
复制代码

  • 查看指定partition
  1. curl -X GET \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic/partitions/0
复制代码

  • 响应
  1. {
  2.         "partition": 0,
  3.         "leader": 0,
  4.         "replicas": [{
  5.                 "broker": 0,
  6.                 "leader": true,
  7.                 "in_sync": true
  8.         }]
  9. }
复制代码

  • 查看指定partition的offset情况
  1. curl -X GET \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic/partitions/0/offsets
复制代码

  • 响应
  1. {
  2.         "beginning_offset": 0,
  3.         "end_offset": 5
  4. }
复制代码
创建bridge consumer


  • 通过bridge消费消息，有个特别且重要的前提：创建bridge consumer，只有先创建了bridge consumer，才能顺利从kafka的broker取到消息
  • 以下命令创建了一个bridge consumer，各参数的含义稍后会说明
  1. curl -X POST http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group \
  2.   -H 'content-type: application/vnd.kafka.v2+json' \
  3.   -d '{
  4.     "name": "bridge-quickstart-consumer",
  5.     "auto.offset.reset": "earliest",
  6.     "format": "json",
  7.     "enable.auto.commit": false,
  8.     "fetch.min.bytes": 16,
  9.     "consumer.request.timeout.ms": 300000
  10.   }'
复制代码

  • 上述请求的参数解释：

  • 对应kafka的group为bridge-quickstart-consumer-group
  • 此bridge consumer的name等于bridge-quickstart-consumer
  • 参数enable.auto.commit表示是否自动提交offset，这里设置成false，表示无需自动提交，后面的操作中会调用API请求来更新offset
  • 参数fetch.min.bytes要特别注意，其值等于16，表示唯有消息内容攒够了16字节，拉取消息的请求才能获取到消息，如果消息内容长度不到16字节，收到的响应body就是空
  • 参数consumer.request.timeout.ms也要注意，这里我设置了300秒，如果超过300秒没有去拉取消息，这个消费者就会被kafka移除（被移除后如果再去拉取消息，kafka会报错：Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the grou）


  • 收到响应如下，instance_id表示这个bridge consumer的身份id，base_uri则是订阅消息时必须使用的请求地址
  1. {
  2.         "instance_id": "bridge-quickstart-consumer",
  3.         "base_uri": "http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
  4. }
复制代码
如何删除bridge consumer


  • 以下命令可以删除consumer，重点是将身份id放入path中
  1. curl -X DELETE http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer
复制代码
订阅指定topic的消息


  • 创建bridge consumer成功后，接下来就能以这个consumer的身份去订阅kafka消息了
  • 执行以下命令可以订阅topic为bridge-quickstart-topic的kafka消息，注意请求地址就是前面创建bridge consumer时返回的base_uri字段
  1. curl -X POST http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/subscription \
  2.   -H 'content-type: application/vnd.kafka.v2+json' \
  3.   -d '{
  4.     "topics": [
  5.         "bridge-quickstart-topic"
  6.     ]
  7. }'
复制代码

  • 从上述请求body可以看出，此请求可以一次订阅多个topic，而且还可以使用topic_pattern（正则表达式）的形式来一次订阅多个topic
  • 订阅完成后，接下来就能主动拉取消息了
拉取消息


  • 在拉取消息之前，请确保已经提前生产了消息
  • 执行以下命令拉取一条消息
  1. curl -X GET http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
  2.   -H 'accept: application/vnd.kafka.json.v2+json'
复制代码

  • 然而，当您执行了上述命令后，会发现返回body为空，别担心，这是正常的现象，按照官方的说法，拉取到的第一条消息就是空的，这是因为拉取操作出触发了rebalancing逻辑（rebalancing是kafka的概览，是处理多个partition消费的操作），再次执行上述命令去拉取消息，这下正常了，body如下
  1. [
  2.         {
  3.                 "topic": "bridge-quickstart-topic",
  4.                 "key": "my-key",
  5.                 "value": "sales-lead-0001",
  6.                 "partition": 0,
  7.                 "offset": 0
  8.         }, {
  9.                 "topic": "bridge-quickstart-topic",
  10.                 "key": "my-key",
  11.                 "value": "sales-lead-0001",
  12.                 "partition": 0,
  13.                 "offset": 1
  14.         }
  15. ]
复制代码
提交offset


  • 前面在创建bridge consumer的时候，参数enable.auto.commit的值等于fasle，表示由调用方主动提交offset到kafka，因此在拉取到消息之后，需要手动更新kafka consumer的offset
  1. curl -X POST http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/offsets
复制代码

  • 该请求无返回body，只要返回码是204就表示成功
设定offset


  • 试想这样的场景：共生产了100条消息，消费者也已经将这100条全部消费完毕，现在由于某种原因，需要从91条开始，重新消费91-100这10条消息（例如需要重新计算），此时可以主动设定offset
  • 先执行以下命令，生产一条消息
  1. curl -X POST \
  2.   http://42.193.162.141:31331/topics/bridge-quickstart-topic \
  3.   -H 'content-type: application/vnd.kafka.json.v2+json' \
  4.   -d '{
  5.     "records": [
  6.         {
  7.             "value": "sales-lead-a002-01234567890123456789",
  8.             "partition": 2
  9.         }
  10.     ]
  11. }'
复制代码

  • 如下图红色箭头，可见当前partition已经生产了75条消息了

  • 咱们先拉取消息，将消息都消费掉

  • 由于没有新生产消息，此时再拉去应该拉取不到了
  • 现在执行以下请求，就可以将offset设置到74
  1. curl -X POST http://42.193.162.141:31331/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions \
  2.   -H 'content-type: application/vnd.kafka.v2+json' \
  3.   -d '{
  4.     "offsets": [
  5.         {
  6.             "topic": "bridge-quickstart-topic",
  7.             "partition": 2,
  8.             "offset": 74
  9.         }
  10.     ]
  11. }'
复制代码

  • 再次拉取消息，发现74和之后的所有消息都可以拉去到了（注意，包含了74）

  • 至此，咱们对生产和发送消息的常用接口都已经操作了一遍，对于常规的业务场景已经够用，接下来的文章，咱们以此为基础，玩出更多花样来
欢迎关注博客园：程序员欣宸

学习路上，你不孤单，欣宸原创一路相伴...

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

缠丝猫

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表