JavaWeb项目架构之Kafka分布式日志队列-低调大师

JavaWeb项目架构之Kafka分布式日志队列

2018-02-05 428

架构、分布式、日志队列，标题自己都看着唬人，其实就是一个日志收集的功能，只不过中间加了一个Kafka做消息队列罢了。

kafka介绍

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。

特性

Kafka是一种高吞吐量的分布式发布订阅消息系统，有如下特性：

通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
支持通过Kafka服务器和消费机集群来分区消息。
支持Hadoop并行数据加载。

主要功能

发布和订阅消息流，这个功能类似于消息队列，这也是kafka归类为消息队列框架的原因
以容错的方式记录消息流，kafka以文件的方式来存储消息流
可以再消息发布的时候进行处理

使用场景

在系统或应用程序之间构建可靠的用于传输实时数据的管道，消息队列功能
构建实时的流数据处理程序来变换或处理数据流，数据处理功能

消息传输流程

Kafka安装

环境

Linux、JDK、Zookeeper

下载二进制程序

wget https://archive.apache.org/dist/kafka/0.10.0.1/kafka_2.11-0.10.0.1.tgz

安装

tar -zxvf kafka_2.11-0.10.0.1.tgz
cd kafka_2.11-0.10.0.1

目录说明

bin 启动,停止等命令
config 配置文件
libs 类库

参数说明

#########################参数解释##############################

broker.id=0  #当前机器在集群中的唯一标识，和zookeeper的myid性质一样

port=9092 #当前kafka对外提供服务的端口默认是9092

host.name=192.168.1.170 #这个参数默认是关闭的

num.network.threads=3 #这个是borker进行网络处理的线程数

num.io.threads=8 #这个是borker进行I/O处理的线程数

log.dirs=/opt/kafka/kafkalogs/ #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个

socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能

socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘

socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小

num.partitions=1 #默认的分区数，一个topic默认1个分区数

log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天

message.max.byte=5242880  #消息保存的最大值5M

default.replication.factor=2  #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务

replica.fetch.max.bytes=5242880  #取消息的最大直接数

log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件

log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除

log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能

zookeeper.connect=192.168.1.180:12181,192.168.1.181:12181,192.168.1.182:1218 #设置zookeeper的连接端口、如果非集群配置一个地址即可

#########################参数解释##############################

启动kafka

启动kafka之前要启动相应的zookeeper集群、自行安装，这里不做说明。

#进入到kafka的bin目录
./kafka-server-start.sh -daemon ../config/server.properties

Kafka集成

环境

spring-boot、elasticsearch、kafka

pom.xml引入：

<!-- kafka 消息队列 -->
<dependency>
<groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>1.1.1.RELEASE</version>
</dependency>

生产者

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
/**
 * 生产者
 * 创建者 科帮网
 * 创建时间    2018年2月4日
 */
@Configuration
@EnableKafka
public class KafkaProducerConfig {

    @Value("${kafka.producer.servers}")
    private String servers;
    @Value("${kafka.producer.retries}")
    private int retries;
    @Value("${kafka.producer.batch.size}")
    private int batchSize;
    @Value("${kafka.producer.linger}")
    private int linger;
    @Value("${kafka.producer.buffer.memory}")
    private int bufferMemory;


    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }
}

消费者

mport java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
/**
 * 消费者
 * 创建者 科帮网
 * 创建时间    2018年2月4日
 */
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
    private int concurrency;
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(1500);
        return factory;
    }

    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }


    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }

    @Bean
    public Listener listener() {
        return new Listener();
    }
}

日志监听

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.itstyle.es.common.utils.JsonMapper;
import com.itstyle.es.log.entity.SysLogs;
import com.itstyle.es.log.repository.ElasticLogRepository;
/**
 * 扫描监听
 * 创建者 科帮网
 * 创建时间    2018年2月4日
 */
@Component
public class Listener {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    
    @Autowired
    private  ElasticLogRepository elasticLogRepository;
    
    @KafkaListener(topics = {"itstyle"})
    public void listen(ConsumerRecord<?, ?> record) {
        logger.info("kafka的key: " + record.key());
        logger.info("kafka的value: " + record.value());
        if(record.key().equals("itstyle_log")){
            try {
                SysLogs log = JsonMapper.fromJsonString(record.value().toString(), SysLogs.class);
                logger.info("kafka保存日志: " + log.getUsername());
                elasticLogRepository.save(log);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

测试日志传输

  /**
    * kafka 日志队列测试接口
    */
   @GetMapping(value="kafkaLog")
   public @ResponseBody String kafkaLog() {
        SysLogs log = new SysLogs();
        log.setUsername("红薯");
        log.setOperation("开源中国社区");
        log.setMethod("com.itstyle.es.log.controller.kafkaLog()");
        log.setIp("192.168.1.80");
        log.setGmtCreate(new Timestamp(new Date().getTime()));
        log.setExceptionDetail("开源中国社区");
        log.setParams("{'name':'码云','type':'开源'}");
        log.setDeviceType((short)1);
        log.setPlatFrom((short)1);
        log.setLogType((short)1);
        log.setDeviceType((short)1);
        log.setId((long)200000);
        log.setUserId((long)1);
        log.setTime((long)1);
        //模拟日志队列实现
        String json = JsonMapper.toJsonString(log);
        kafkaTemplate.send("itstyle", "itstyle_log",json);
        return "success";
   }

Kafka与Redis

之前简单的介绍过，JavaWeb项目架构之Redis分布式日志队列，有小伙伴们聊到， Redis PUB/SUB没有任何可靠性保障，也不会持久化。当然了，原项目中仅仅是记录日志，并不是十分重要的信息，可以有一定程度上的丢失

Kafka与Redis PUB/SUB之间最大的区别在于Kafka是一个完整的分布式发布订阅消息系统，而Redis PUB/SUB只是一个组件而已。

使用场景

Redis PUB/SUB
消息持久性需求不高、吞吐量要求不高、可以忍受数据丢失
Kafka
高可用、高吞吐、持久性、多样化的消费处理模型

开源项目源码(参考)：https://gitee.com/52itstyle/spring-boot-elasticsearch

作者：小柒

出处： https://blog.52itstyle.com

分享是快乐的，也见证了个人成长历程，文章大多都是工作经验总结以及平时学习积累，基于自身认知不足之处在所难免，也请大家指正，共同进步。

微信关注我们

原文链接：https://yq.aliyun.com/articles/435779

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Android平台的代码热更新

跨平台和多语言支持 Unity之所以能跨平台发布，依赖于其底层运行的是Mono。Mono可以简单地理解为开源版本的.Net框架(基于C#和CLI的ECMA标准)，当微软开源了.Net的核心库后，Mono也在做相应的改善和优化。Unity基于Mono支持三种脚本语言：C#、Boo和UnityScript，其中 UnityScript是JavaScript的强类型的变种，开发人员使用的最多的还是C#和UnityScript。这里的脚本并不是传统意义上的脚本即解释执行，而是将代码编译成IL后在Mono的Runtime上运行(类似于Java的运行原理)，因此Unity并不支持传统意义的游戏脚本编程。而iOS 平台的IL2CPP的出现，意味着Unity中游戏编程语言的静态化的发展，所以目前我们推测Unity不会原生支持动态语言(Lua)。为何不能代码热更新 1. 资源和脚本对于Stealth项目中的一个Prefab文件“door_generic_slide.prefab”，其在Inspector中的详情如下：可以看出此Prefab有6个组件，其中1个组件是脚本，且脚本中含有可序列化的变量(...

2018-02-05

748

html代码： <!DOCTYPE html> <html lang="en"> <head> <meta http-equiv="content-type" content="text/html; charset=utf-8" /> <title>Image Gallery</title> <script type="text/javascript" src="scripts/showPic.js"></script> <link rel="stylesheet" href="styles/layout.css" type="text/css" media="screen" /> </head> <body> <h1>Snapshots</h1> <ul id="imagegallery"> <li> <a href="images/fireworks.jpg" title="A fireworks ...

2018-02-05

672

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。