安科网

2018年第28周-Kafka环境搭建和其Java例子

jiaomrswang

jiaomrswang

2019-06-27

关注 关注

安装kafka

下载

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz

配置文件

我部署在服务器

s1.jevoncode.com
s2.jevoncode.com
s3.jevoncode.com

都在目录/mydata1/kafka_2.11-1.1.0
修改config/server.properties文件的broker.id和zk，broker.id要集群上唯一

broker.id=1
zookeeper.connect=s1.jevoncode.com:2181,s2.jevoncode.com:2181,s3.jevoncode.com:2181
#如只能让内网访问，则需配置这个
listeners=PLAINTEXT://192.168.56.4:9092

启动kafka

先在每个服务器启动zookeeper

zkServer.sh start

再在每个服务器启动以下命令

nohup ./bin/kafka-server-start.sh config/server.properties &

Kafka的java例子

项目配置

项目使用spring-boot和和spring-kafka，pom文件如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.jc.demo</groupId>
    <artifactId>jc-demo-kafka</artifactId>
    <version>0.0.1.0-SNAPSHOT</version>
    <packaging>jar</packaging>
    <name>oneapp-archetype-test</name>
    <url>http://www.jevoncode.com</url>

    <properties>
        <!-- Every File in Project Enconding -->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

        <!-- Compiling Time Enconding -->
        <maven.compiler.encoding>UTF-8</maven.compiler.encoding>

        <!-- Compiling Time JDK Version -->
        <java.version>1.7</java.version>

        <!-- Test -->
        <junit.version>4.12</junit.version>


        <!-- Logging -->
        <slf4j.version>1.7.21</slf4j.version>
        <logback.version>1.1.7</logback.version>


    </properties>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.3.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>


    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- http://kafka.apache.org/documentation.html#producerapi-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.1.0</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        <!-- Log依赖 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!-- logback -->
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>${logback.version}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
            <version>${logback.version}</version>
        </dependency>

    </dependencies>

</project>

配置文件

application.properties

#kafka地址
jc.kaHost=s1.jevoncode.com:9092,s2.jevoncode.com:9092,s3.jevoncode.com:9092

Topic配置

package com.jc.demo.springboot.config;

import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
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.KafkaAdmin;

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

/**
 * 创建kafka的topic
 * 如果kafka不存在此topic则会自动创建，存在则不改变kafka的topic
 */
@Configuration
@EnableKafka
public class TopicConfig {


    public static final String TOPIC_JC_KAFKA_DEMO = "jc-demo-kafka";

    @Value("${jc.kaHost}")
    String kaHost;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kaHost);
        return new KafkaAdmin(configs);
    }

    @Bean
    public NewTopic foo() {
        //第一个是参数是topic名字，第二个参数是分区个数，第三个是topic的复制因子个数
        //当broker个数为1个时会创建topic失败，
        //提示：replication factor: 2 larger than available brokers: 1
        //只有在集群中才能使用kafka的备份功能
        //以kafka的分区机制来说，我将其numParitions个数设置为broker个数，也就是3
        return new NewTopic(TOPIC_JC_KAFKA_DEMO, 3, (short) 2);
    }

//
//    @Bean
//    public NewTopic topic1(){
//        return new NewTopic("jc-demo-kafka2", 10, (short) 2);
//    }
}

启动类

package com.jc.demo.springboot;


import com.jc.demo.springboot.config.ApplicationConfig;
import com.jc.demo.springboot.service.DoService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;

@SpringBootApplication
@Import({ApplicationConfig.class})
@PropertySource("classpath:application.properties")
public class DemoApplication implements CommandLineRunner{

    @Autowired
    private DoService doService;

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }


    @Override
    public void run(String... args) throws Exception {
        doService.HelloWorld();
    }
}

生产者

配置类KafkaProducerConfig

package com.jc.demo.springboot.config;

import com.jc.demo.springboot.service.MyListener;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
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.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;

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

@Configuration
@EnableKafka
public class KafkaProducerConfig {

    @Value("${jc.kaHost}")
    String kaHost;


    /* --------------producer configuration-----------------**/
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kaHost);
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

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


    /* --------------kafka template configuration-----------------**/
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        return kafkaTemplate;
    }

}

服务接口和类（生产发送消息）

DoService接口

package com.jc.demo.springboot.service;

public interface DoService {

    void HelloWorld();
}

DoService实现类，调用kafkaTemplate完成发送消息到kafka

package com.jc.demo.springboot.service;

import com.jc.demo.springboot.config.TopicConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigInteger;

@Service
public class DoServiceImpl implements  DoService {
    @Autowired
    KafkaTemplate kafkaTemplate;

    @Override
    public void HelloWorld() {
        String phone = "18689206965";
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            kafkaTemplate.send(TopicConfig.TOPIC_JC_KAFKA_DEMO, partition(phone), phone, "jevoncode" + System.currentTimeMillis());
        }
    }


    /**
     * 根据手机号计算分区
     * @return
     */
    private int partition(String phone) {
        int hash = phone.hashCode();
        int partition = new BigInteger(Integer.toString(hash)).mod(new BigInteger("3")).intValue();     //由于总共有3个分区，所以得去3的模
        System.out.println(partition);
        return partition;
    }

}

此时执行DemoApplication的main方法，就可以生产一个字符串"jevoncode"到kafka的0分区上。
可以使用命令查看：

[jevoncode@s1 kafka_2.11-1.1.0]$ ./bin/kafka-console-consumer.sh --bootstrap-server s1.jevoncode.com:9092 --topic jc-demo-kafka --from-beginning
jevoncode1531654603522
jevoncode1531654689283
jevoncode1531654690331
jevoncode1531654691332
jevoncode1531654692332
....

消费者

此时application.properties配置文件需增加消费者配置

#kafka地址
jc.kaHost=s1.jevoncode.com:9092,s2.jevoncode.com:9092,s3.jevoncode.com:9092

#############以下是消费者端的配置###########################
#kafka消费者 groupId配置
jc.consumer.group.id=jc-consumer-group-1
#kafka消费者 分区配置，这样就可以指定每个消费者所消费的分区，提高吞吐量
jc.consumer.partitions=0,1,2
#一次从kafka拉的最大消息数
jc.max.poll.records=100

消费者配置类

package com.jc.demo.springboot.config;

import com.jc.demo.springboot.listener.KafkaListener;
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.core.*;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.BatchLoggingErrorHandler;

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

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Value("${jc.kaHost}")
    String kaHost;
    /**
     * 批量拉去消息的个数
     */
    @Value("${jc.max.poll.records}")
    String maxPool;



    /* --------------consumer configuration-----------------**/
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kaHost);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "0");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPool);
        return props;
    }


    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties().setBatchErrorHandler(new BatchLoggingErrorHandler());
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        return factory;
    }

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

    @Bean//消息监听器
    public KafkaListener kafkaListener() {
        return new KafkaListener();
    }

}

kafka消费监听类（接收和处理消息）

package com.jc.demo.springboot.listener;

import com.jc.demo.springboot.config.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

import java.util.List;

public class KafkaListener {
    private static Logger logger = LoggerFactory.getLogger(KafkaListener.class);
    @org.springframework.kafka.annotation.KafkaListener(id = "${jc.consumer.group.id}",topicPartitions =//配置topic和分区
                    { @TopicPartition(topic = TopicConfig.TOPIC_JC_KAFKA_DEMO, partitions ="#{'${jc.consumer.partitions}'.split(',')}")})
    public void receive(@Payload List<String> messages,
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                         @Header(KafkaHeaders.OFFSET) List<Long> offsets, Acknowledgment ack) {

        for (int i = 0; i < messages.size(); i++) {
            String msg = "message='" + messages.get(i) + "' with partition-offset='" + partitions.get(i) + "-" + offsets.get(i) + "'";

            logger.debug("receive messages {}",msg);
        }
        ack.acknowledge();
        logger.debug("all batch messages {} consumed",messages.size());
    }


}

此时运行kafka就可以DemoApplication就可以看到如下日志：

07-15 19:39:25 [jc-consumer-group-1-0-C-1] DEBUG com.jc.demo.springboot.listener.KafkaListener - receive messages message='jevoncode1531654765409' with partition-offset='0-17'
07-15 19:39:25 [jc-consumer-group-1-0-C-1] DEBUG com.jc.demo.springboot.listener.KafkaListener - all batch messages 1 consumed

kafka kafka命令

jiaomrswang

jiaomrswang

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

.NET Core下使用Kafka的方法步骤

# 修改每个topic的默认分区参数num.partitions，默认是1，具体合适的取值需要根据服务器配置进程确定，UCloud.ukafka = 3. Segment：partition物理上由多个segment组成，下面2.2和2.3有详细说明。pa

Kafka 2020-09-18

SAP Data Intelligence里的Kafka Producer和Consumer

新建一个graph，使用典型的生产者-消费者模型：将Data Generator生成的数据交给kafka Producer operator；而Kafka Consumer从kafka producer里读取出data Generator生成的数据，通过T

guicaizhou 2020-09-15

Windows环境下载安装Kafka

2.需要先启动ZooKeeper服务器(可以通过与kafka打包在一起的便捷脚本来快速简单地创建一个单节点ZooKeeper实例。

jiangkai00 2020-07-18

kafka和zookeeper集群搭建脚本

#3个节点集群使用，分别在节点1 2 3执行该脚本即可。# 重点 Type 必须为 forking 否则无法启动，

CobingLiu 2020-06-16

Unbuntu16搭建Kafka环境总结

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1

yangyutong00 2020-06-12

linux虚拟机中启动服务命令大全以及常见问题解决

进入redis命令 ps -ef | grep redis 查看redis服务是否开启 如果开启可以直接 redis-cli -h 192.168.107.128 -p 6379 进到redis的bin目录下执行 这个是开启redis服

sweetgirl0 2020-06-09

架构师的选择，Pulsar还是Kafka？

最近，我一直在研究Pulsar及其与Kafka的比较。快速搜索将显示两个最著名的开源消息传递系统之间存在当前的"战争"。作为Kafka的用户，我确实对Kafka的某些问题感到困惑，并且我对Pulsar感到非常失望。在本文中，我将重点介绍

yanghuashuiyue 2020-11-14

Kafka之消费与心跳

导读kafka是一个分布式，分区的，多副本的，多订阅者的消息发布订阅系统，可以用于搜索日志，监控日志，访问日志等。今天小编来领大家一起来学习一下Kafka消费与心跳机制。Kafka提供了非常简单的消费API，使用者只需初始化Kafka的Broker Se

liuxingen 2020-11-13

刨根问底,Kafka消息中间件到底会不会丢消息

本文转载自微信公众号「爱笑的架构师」，作者雷架。大型互联网公司一般都会要求消息传递最大限度的不丢失，比如用户服务给代金券服务发送一个消息，如果消息丢失会造成用户未收到应得的代金券，最终用户会投诉。为避免上面类似情况的发生，除了做好补偿措施，更应该在系设计的

wangying 2020-11-13

记一次生产 Kafka 挂掉的那几分钟

本文转载自微信公众号「Java极客技术」，作者鸭血粉丝。Hello，大家好，我是阿粉，作为一个后端工程师不经历几次生产事故怎么能成长!阿粉工作几年来，大大小小，重要不重要的事故也经历了不少，有损失几十万的，有对业务毫无影响但是不应该发生的，每一次事故都是一

王谦 2020-11-03

面试 | 再有人问Kafka，别说你不会！

Kafka 是一个优秀的分布式消息中间件，许多系统中都会使用到 Kafka 来做消息通信。对分布式消息系统的了解和使用几乎成为一个后台开发人员必备的技能。今天就从常见的 Kafka 面试题入手和大家聊聊 Kafka 的那些事儿。生产者负责创建消息，然后将其

huangwei00 2020-10-14

RabbitMQ七战Kafka，差异立现！

作为一个有丰富经验的微服务系统架构师，经常有人问我，“应该选择RabbitMQ还是Kafka？”。基于某些原因， 许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有

shenzhenzsw 2020-10-09

Kafka突然宕机了？稳住，莫慌！

问题要从一次 Kafka 的宕机开始说起。笔者所在的是一家金融科技公司，但公司内部并没有采用在金融支付领域更为流行的 RabbitMQ，而是采用了设计之初就为日志处理而生的 Kafka，所以我一直很好奇 Kafka 的高可用实现和保障。从 Kafka 部

guicaizhou 2020-09-30

监控kafka队列长度

#!df=pd.DataFrame(columnList,columns=["topics","LAG"]). dfResult=df.groupby("topics",as_index=Fals

jiaomrswang 2020-09-23

Kafka都没整明白，还敢去面试？

放眼当下数据为王的时代，深入了解 Apache Kafka 及其常见的部署应用，快速实现数据架构已是大势所趋，刻不容缓。以下分别 Kafka 架构，四大核心 API，典型应用场景，Kafka 代理与消息主题，集群的创建，流 APIs及其处理模式等不同方面展

jyj0 2020-09-21

kafka技术题

将向 Kafka topic 发布消息的程序成为 producers.将预订 topics 并消费消息的程序成为 consumer.Kafka 以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个 broker.producers 通过网络将消息发

hannuotayouxi 2020-08-20

kafka常见面试题

我们举个例子说明下运维中面对的复杂性，我们都知道 kafka 有个 ISR集合，我先说明下这个概念：

yangyutong00 2020-08-09

Windows 安装kafka

直接去 D:\Tool\zookeeper-3.4.14\bin， 运行 zkServer.cmd. kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1

amwayy 2020-08-03

【kafka】消费组消费主题

当消费组中的消费者数量大于分区数量时，会存在浪费。因为有消费者分配不到partition进行消费。在消费者不饱和的情况下，一个消费者是可以去消费多个分区的数据的 。生产者生产投递9条消息，可以9条消息被投递到3不同分区，然后三个分区对应三个不同的消费者进行

yangyutong00 2020-08-01

logstash_output_kafka:Mysql同步Kafka深入详解

实际业务场景中，会遇到基础数据存在Mysql中，实时写入数据量比较大的情景。迁移至kafka是一种比较好的业务选型方案。如果需要同步历史全量数据+实时更新数据，建议使用logstash。实际上，核心logstash的同步原理的掌握，有助于大家理解类似的各种

weikaixxxxxx 2020-08-01
jiaomrswang

jiaomrswang

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号