kafka应用场景有哪些_kafka顺序性的消费

kafka应用场景有哪些_kafka顺序性的消费序在学习一门新技术之前,我们需要先去了解一下这门技术的具体应用场景,使用它能够做什么,能够达到什么目的,学习kafka的初衷是用作消息队列;但是还可以使用KafkaStream进行一些实时的流计算,多用于大数据处理;也可以做日志收集汇总、网站活动跟踪等任务。消息队列kafka可以很好的替代一些传统的消息系统,kafka具有更好的吞吐量,内置的分区使kaf…

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

在学习一门新技术之前,我们需要先去了解一下这门技术的具体应用场景,使用它能够做什么,能够达到什么目的,学习kafka的初衷是用作消息队列;但是还可以使用Kafka Stream进行一些实时的流计算,多用于大数据处理;也可以做日志收集汇总、网站活动跟踪等任务。

消息队列

kafka可以很好的替代一些传统的消息系统,kafka具有更好的吞吐量,内置的分区使kafka具有更好的容错和伸缩性,这些特性使它可以替代传统的消息系统,成为大型消息处理应用的首选方案。

场景:异步、解耦、削峰填谷

  1. 生成订单:给不同的产品业务线分配同一个topic的不同partition,用户下单后根据订单类型发送到对应的partition
  2. 消息通知:用户登录后计算积分
  • 消息生产者

    public static void main(String[] args) throws Exception {
      Properties prop = new Properties();
      prop.put("bootstrap.servers", "127.0.0.1:9092");
      prop.put("acks", "all");
      prop.put("retries", "0");
      // 缓冲区大小
      prop.put("batch.size", "10");
      prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      KafkaProducer<String, String> producer = new KafkaProducer<>(prop);
      for (int i = 0; i < 101; i++) {
        ProducerRecord<String, String> record = new ProducerRecord<>("my_topics", "value_" + i);
        // 阻塞到消息发送完成
        producer.send(record).get();
      }
      // 刷新缓冲区,发送到分区,并清空缓冲区
      // producer.flush();
      // 关闭生产者,会阻塞到缓冲区内的数据发送完
      producer.close();
      // producer.close(Duration.ofMillis(1000));
    }
    

    生产者发送消息是先将消息放到缓冲区,当缓冲区存满之后会自动flush,或者手动调用flush()方法

  • 消息消费者

    public static void main(String[] args) {
      Properties properties = new Properties();
      properties.put("bootstrap.servers", "127.0.0.1:9092");
      properties.put("group.id", "cc_consumer");
      properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
      // 指定topic
      consumer.subscribe(Arrays.asList("my_topics"));
      // 指定topic的partition
      // TopicPartition partition0 = new TopicPartition("my_topics", 10);
      // consumer.assign(Arrays.asList(partition0));
      try {
        while (true) {
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
          for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.toString());
          }
        }
      } finally {
        consumer.close(Duration.ofMillis(2000));
      }
    }
    

流计算

[todo]

日志收集

应用程序的日志可以通过log4j收集日志信息,并将日志直接打到kafka中:客户端—>应用—>kafka

SpringBoot中默认使用的是logback,所以要在引入SpringBoot的jar包时排除掉logback的jar包

日志消息发送有同步和异步两种方式,由KafkaAppender中的syncSend属性决定,默认为true(同步)

> <Kafka name="KAFKA-LOGGER" topic="cc_log_test" syncSend="false">
>
  • pom.xml
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
  <exclusions>
    <exclusion>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-logging</artifactId>
    </exclusion>
  </exclusions>
</dependency>
<!-- springboot 1.3.x之前版本是log4j,之后版本都是log4j2 -->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>
  • log4j2.xml
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="off">
    <Properties>
    </Properties>
    <Appenders>
        <Console name="STDOUT" target="SYSTEM_OUT">
            <PatternLayout pattern="%d %p %c{1.} %t %m%n"/>
        </Console>
	      <!--kafka topic-->
        <Kafka name="KAFKA-LOGGER" topic="my_topics">
          	<!--JsonLayout:日志格式为json,方便在ES中处理-->
            <JsonLayout/>
          	<!--kafka server的ip:port-->
            <Property name="bootstrap.servers">127.0.0.1:9092</Property>
            <Property name="retries">3</Property>
            <Property name="linger.ms">1000</Property>
            <Property name="buffer.memory">10485760</Property>
        </Kafka>
        <Async name="ASYNC-KAFKA-LOGGER">
            <AppenderRef ref="KAFKA-LOGGER"/>
            <LinkedTransferQueue/>
        </Async>
    </Appenders>
    <Loggers>
      	<!--日志级别大于info都会被记录到Kafka-->
        <Logger name="cc.kevinlu.springbootkafka.controller.MessageController" level="info"
                additivity="false">
            <AppenderRef ref="KAFKA-LOGGER"/>
        </Logger>
        <!-- Root表示所有Logger用Root中的Appender打印日志  -->
        <Root level="info">
            <AppenderRef ref="STDOUT"/>
        </Root>
    </Loggers>
</Configuration>
  • code
@GetMapping("/log")
public String sendLog() {
  for (int i = 0; i < 10; i++) {
    log.info("kafka log i = " + i);
  }
  return "success";
}
  • consumer视图

image-20200419032218971

网站活动跟踪

  1. 前端Nodejs控制

    Node接入kafka需要使用kafka-node库,下面是网上的例子

    var kafka = require('kafka-node'),
        Producer = kafka.Producer,
        client = new kafka.KafkaClient({kafkaHost: 'localhost:9092'});
    /**
     * 定义生产类
     * partitionerType 定义
     * 0:默认模式 只产生数据在第一个分区
     * 1:随机分配,在分区个数内,随机产生消息到各分区
     * 2:循环分配,在分区个数内,按顺序循环产生消息到各分区
    */   
    var producerOption = {
        requireAcks: 1,
        ackTimeoutMs: 100,
        partitionerType: 0 //默认为第一个分区
    };
    var producer = new Producer(client,producerOption);
    /**
     * TOPIC的创建需要在命令行进行创建,以便指定分区个数以及备份个数
     * PS:kafka-node的创建topic不行,不能创建分区
     * 产生消息,如果不指定partition
     * 则根据 partitionerType 的值来指定发送数据到哪个分区
     * 我们创建的topic-test-one只有一个分区,所以只能产生数据到第1个分区(下标0),否则不会生产数据
     */
    function getPayloads(){
        return [
            {topic:"topic-test-one",messages:JSON.stringify({"name":"jack","age":"120"}),partition:0}
        ];
    }
    
    producer.on("ready",function(){
        setInterval(function(){
            producer.send(getPayloads(),function(err,data){
                if(!err){
                    console.log("send message complete!data:"+JSON.stringify(data),new Date());
                }
            });
         },1000);
    });
    
    producer.on('error', function (err) {console.log("send message error!\r\n"+err);})
    
  2. 后端日志控制

    后端也可以使用log4j的日志系统来完成,拦截所有需要监控的api请求,使用log4j输出日志到kafka队列中,和上述日志收集方法相同。若同一个应用中需要通过日志输出到kafka的多个topic中,可以使用log4j的Marker标记来区分,配置如下:

    <?xml version="1.0" encoding="UTF-8"?>
    <Configuration status="off">
        <Properties>
        </Properties>
        <Appenders>
            <Console name="STDOUT" target="SYSTEM_OUT">
                <PatternLayout pattern="%d %p %c{1.} %t %m%n"/>
            </Console>
          	<!-- 日志收集 -->
            <Kafka name="KAFKA-LOGGER" topic="cc_log_test" syncSend="false">
                <JsonLayout/>
                <Property name="bootstrap.servers">127.0.0.1:9092</Property>
                <Property name="retries">3</Property>
                <Property name="linger.ms">1000</Property>
                <Property name="buffer.memory">10485760</Property>
                <Filters>
                  	<!-- 通过Marker过滤消息 -->
                    <MarkerFilter marker="Kafka" onMatch="ACCEPT" onMismatch="DENY"/>
                </Filters>
            </Kafka>
          	<!-- 轨迹跟踪 -->
            <Kafka name="KAFKA-TRACK-LOGGER" topic="cc_test1" syncSend="false">
                <JsonLayout/>
                <Property name="bootstrap.servers">127.0.0.1:9092</Property>
                <Property name="retries">3</Property>
                <Property name="linger.ms">1000</Property>
                <Property name="buffer.memory">10485760</Property>
                <Filters>
                  	<!-- 通过Marker过滤消息 -->
                    <MarkerFilter marker="Track" onMatch="ACCEPT" onMismatch="DENY"/>
                </Filters>
            </Kafka>
            <Async name="ASYNC-KAFKA-LOGGER">
                <AppenderRef ref="KAFKA-LOGGER"/>
                <AppenderRef ref="KAFKA-TRACK-LOGGER"/>
                <LinkedTransferQueue/>
            </Async>
        </Appenders>
        <Loggers>
            <Logger name="cc.kevinlu.springbootkafka.controller" level="info"
                    additivity="false">
                <AppenderRef ref="KAFKA-LOGGER"/>
                <AppenderRef ref="KAFKA-TRACK-LOGGER"/>
            </Logger>
            <Root level="info">
                <AppenderRef ref="STDOUT"/>
            </Root>
        </Loggers>
    </Configuration>
    
    private final static Marker KAFKA_MARKER       = MarkerManager.getMarker("Kafka");
    private final static Marker KAFKA_TRACK_MARKER = MarkerManager.getMarker("Track");
    
    @GetMapping("/log")
    public String sendLog() {
      // 轨迹跟踪
      log.info(KAFKA_TRACK_MARKER, "send async message!");
      for (int i = 0; i < 10; i++) {
        // 日志收集
        log.info(KAFKA_MARKER, "kafka log i = {}", i);
      }
      return "success";
    }
    
  3. 前端+后端组合

    后端提供API供前端传递轨迹,后端接收到请求之后将消息同步到kafka中。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/181994.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • mybatisCodeHelpe离线激活码【2022.01最新】

    (mybatisCodeHelpe离线激活码)JetBrains旗下有多款编译器工具(如:IntelliJ、WebStorm、PyCharm等)在各编程领域几乎都占据了垄断地位。建立在开源IntelliJ平台之上,过去15年以来,JetBrains一直在不断发展和完善这个平台。这个平台可以针对您的开发工作流进行微调并且能够提供…

  • 快速排序法——quicksort in java

    快速排序法——quicksort in java

  • 树莓派3B+ 安装系统

    树莓派3B+ 安装系统树莓派3B+安装系统对于树莓派3B+系统安装方法有很多,我就介绍比较普通的一种。适合小白操作!安装概要步骤:官网下载系统-》刷入TF卡-》设置开启显示器和SSH-》通电-》进入系统详细步骤进入官方网站下载系统镜像。https://www.raspberrypi.org/downloads/来到下面这个页面如果感觉下载速度慢,可以将下载链接放到迅雷里面下,基本可以做到满速下载!下载

  • zookeeper的基本操作_奇门遁甲实战入门之五

    zookeeper的基本操作_奇门遁甲实战入门之五本原创入门教程,涵盖ZooKeeper核心内容,通过实例和大量图表,结合实战,帮助学习者理解和运用,任何问题欢迎留言。目录:zookeeper介绍与核心概念 安装和使用 ZooKeeper分布式锁实现 ZooKeeper框架Curator分布式锁实现及源代码分析 zookeeper开发实战(java客户端)本章是后续学习的基石,只有充分理解了分布式系统的概念和面临的问题,以及Z…

  • 观察者模式是非常常用的设计模式_实现一个观察者模式

    观察者模式是非常常用的设计模式_实现一个观察者模式好久没有写博客啦,之前看完了《设计模式之禅》也没有总结一下,现在回忆一下设计模式之观察者模式。1.什么是观察者模式简单情形:有A、B、C、D等四个独立的对象,其中B、C、D这三个对象想在A对象发生改

  • 三分钟教你快速选择机器视觉传感器

    三分钟教你快速选择机器视觉传感器点击上方“小白学视觉”,选择加"星标"或“置顶”重磅干货,第一时间送达本文转自|新机器视觉六招教您合理选择机器视觉传感器传感器是一种检测装置,能感受到被测量的信息,并能…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号