Kafka FAQ

标签：-- FAQ RabbitMQ kafka 消息 Pulsar Kafka

1. 首先在docker上安装kafka和zookeeper, 加起来要1G的image size, 先安装1台,测试完后再加集群对比测试性能

用spring-kafka的包写个测试例子

private KafkaTemplate<String,Object> kafkaTemplate;

public boolean send(@RequestParam String message){
        kafkaTemplate.send("testTopic",message);
}

连接别的机器上的kafka,出现下面错误

Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

但application.yml配置的不是本地127.0.0.1地址, 这个就要改kafka的 advertised.listeners

# 允许外部端口连接                                            
listeners=PLAINTEXT://0.0.0.0:9092  
# 外部代理地址                                                
advertised.listeners=PLAINTEXT://121.201.64.12:9092

Kafka连接服务器出现:Connection to node 1 (localhost/127.0.0.1:9092) could not be established

【kafka】查看消息情况

docker exec -it [docker container id] bash 进入docker 容器的控制台
cd /opt/bitnami/kafka/bin/

命令格式：kafka-consumer-groups.sh --bootstrap-server IP:端口,IP:端口 --describe --group 组名

例如：./kafka-consumer-groups.sh --bootstrap-server 192.168.110.97:9092 --group group --describe

RabbitMQ 自带一个管理界面, Kafka就比较原始了. 下面比较一下 RabbitMQ和Kafka:

Kafka支持消息重播. 假如要接入FLink,就不能用RabbitMQ
RabbitMQ的优势在于能够灵活地路由消息,Kafka不支持路由。 Kafka主题分为多个分区，每个分区包含不可更改的消息
RabbitMQ支持优先级队列,Kafka中，不能以优先级顺序发送消息。 Kafka中的所有消息都按照接收顺序存储和传递。
RabbitMQ采用Push 推送消息, Kafka采用Pull请求, 消费者从给定偏移量中请求一批消息

再比较一下 Pulsar和Kafka:

plusar扩容，增加节点时比kafka更快更方便,P而Puslar是对数据分片，容易扩展 ;Kafka的topic的性能扩展受限于partitions的个数
流量负载均衡. Kafka要用商业工具,手动操作.比较麻烦(需求迁移分区,相当增加副本)
Pulsar分层存储，您可以自动将旧消息推入实际上无限，廉价的云存储. Kafka现在Broker和存储是一体的

选择Pulsar而不是Kafka的5个原因

Kafka Improvement Proposals - Apache Kafka - Apache Software Foundation

新浪微博从 Kafka 到 Pulsar 的演变

假设每日有1000w条消息, 每条消息10kB, 消息储存7天,用不同的消息队列,价格是多少呢?

消息队列 Pulsar 版价格总览按腾讯云这个价格 API 调用1200*2=2400元,存储价格7TB=7000G*0.42/周*3 副本=8820, 主题每月不能超2000个,不然另外收费. 毛估一下每月1万2打住了.

阿里云kafka 计算下来1个月1万左右, 感觉2者价格差不多

消息队列 RabbitMQ 版包年包月, 每日有1000w条消息折算下面每秒才115条消息*2=230, 选用TPS 10000/秒就应该绰绰有余了, 一个月才1300,便宜好多啊, 不过阿里云最高TPS才5000/秒, 超过这个数量的还是选择Pulsar了吧.

用kafkaTemplate 一个连接连续发送10w条记录,用8个线程发只需要1.6秒.在发送的同时接收，10w条记录4秒，速度真的挺快的哦!!!

多线程发送，消息是乱序的，假如要有次序，只用1个线程发，3.6秒

在console统计不了那么多，要写log，有2个配置文件要配一下

logging:
  #level 日志等级 指定命名空间的日志输出
  level:
    com.fishpro.kafka: debug   #根据包名修改
  #file 指定输出文件的存储路径
  #file: logs/app.log
  #pattern 指定输出场景的日志输出格式
  pattern:
    console: "%d %-5level %logger : %msg%n"
    file: "%d %-5level [%thread] %logger : %msg%n"

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    <contextName>logback</contextName>
    <!--输出到控制台-->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <!--按天生成日志-->
    <appender name="logFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <Prudent>true</Prudent>
        <!-- 过滤器，只打印ERROR级别的日志 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件输出的文件名-->
            <FileNamePattern>
                applog/%d{yyyy-MM-dd}/%d{yyyy-MM-dd}.log
            </FileNamePattern>
            <!--日志文件保留天数-->
            <MaxHistory>15</MaxHistory>
        </rollingPolicy>
        <layout class="ch.qos.logback.classic.PatternLayout">
            <Pattern>
                %d{yyyy-MM-dd HH:mm:ss} -%msg%n
            </Pattern>
        </layout>
    </appender>

    <logger name="com.fishpro.kafka" additivity="false">
        <appender-ref ref="console"/>
        <appender-ref ref="logFile"/>
    </logger>

    <!-- 设置Spring&Hibernate日志输出级别 -->
    <logger name="org.springframework" level="WARN"/>
    <logger name="org.mybatis" level="WARN"/> 
    <logger name="com.ibatis" level="DEBUG"/> 
    <logger name="com.ibatis.common.jdbc.SimpleDataSource" level="DEBUG"/> 
    <logger name="com.ibatis.common.jdbc.ScriptRunner" level="DEBUG"/> 
    <logger name="com.ibatis.sqlmap.engine.impl.SqlMapClientDelegate" level="DEBUG"/>


       
    <logger name="java.sql.Connection" level="DEBUG"/>  
    <logger name="java.sql.Statement" level="DEBUG"/> 
    <logger name="java.sql.PreparedStatement" level="DEBUG"/> 
    <logger name="com.ruidou.baoqian.mapper" level="DEBUG"/>

    <!-- 开发环境下的日志配置 -->
    <root level="error">
        <appender-ref ref="console"/>
        <appender-ref ref="logFile"/>
    </root>
</configuration>

logback-spring.xml

如果10w个连接,每个连接发一条消息呢?

kafka如何限制连接数呢?

如何保证kafka已经接收到消息了呢?

标签：--,FAQ,RabbitMQ,kafka,消息,Pulsar,Kafka
From： https://www.cnblogs.com/zitjubiz/p/16592110.html

【kafka】查看消息情况

相关文章

赞助商

阅读排行