0.8.2kafka集群配置

Posted 2022-11-26 正义飞

hadoop,spark,kafka交流群：224209501

1，kafka配置

kafka官网的配置参数基本已经满足大家的需求了。

# Replication configurations
num.replica.fetchers=4
replica.fetch.max.bytes=1048576
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.socket.timeout.ms=30000
replica.socket.receive.buffer.bytes=65536
replica.lag.time.max.ms=10000
replica.lag.max.messages=4000

controller.socket.timeout.ms=30000
controller.message.queue.size=10

# Log configuration
num.partitions=8
message.max.bytes=1000000
auto.create.topics.enable=true
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.hours=168
log.flush.interval.ms=10000
log.flush.interval.messages=20000
log.flush.scheduler.interval.ms=2000
log.roll.hours=168
log.retention.check.interval.ms=300000
log.segment.bytes=1073741824

# ZK configuration
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000

# Socket server configuration
num.io.threads=8
num.network.threads=8
socket.request.max.bytes=104857600
socket.receive.buffer.bytes=1048576
socket.send.buffer.bytes=1048576
queued.max.requests=16
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100

kafka集群配置

1，修该broker id

broker.id [阿拉伯数字，集群中不能重复]

2，zookeeper的地址列表

zookeeper.connect=ukafka-rmicyl-1-bj04.service:2181,ukafka-rmicyl-2-bj04.service:2181,ukafka-rmicyl-3-bj04.service:2181

3,广播参数

假如broker所在为内网环境并经过nat，consumer和producer所在网络在nat外部，需要配置这两个参数为nat网关ip和响应端口。kafka目前还不支持，内外网(consumer和producer一个在内网一个在外网)。

# Hostname the broker will advertise to producers and consumers. If not set, it uses the
# value for "host.name" if configured.  Otherwise, it will use the value returned from
# java.net.InetAddress.getCanonicalHostName().
advertised.host.name=<hostname routable by clients>
# The port to publish to ZooKeeper for clients to use. If this is not set,
# it will publish the same port that the broker binds to.
advertised.port=<port accessible by clients>

Kafka 0.10.0 这个版本参数如下：

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = security_protocol://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://your.host.name:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://your.host.name:9092

kafka的生产者demo

props.put("metadata.broker.list", "ip1:9092,ip2:9092,ip3:9092");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("batch.size", "307200");
props.put("linger.ms", "10");
props.put("producer.type", "async");
config = new ProducerConfig(props);
producer = new Producer<String, String>(config);
String key=hell0;
String value=hi;
for(int i=0;i<10i++)
KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, key+Integer.toString(i), value+Integer.toString(i));
producer.send(msg);

消费者demo

import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConsumerGroupExample 
    private final ConsumerConnector consumer;
    private final String topic;
    private  ExecutorService executor;

    public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) 
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig(a_zookeeper, a_groupId));
        this.topic = a_topic;
    

    public void shutdown() 
        if (consumer != null) consumer.shutdown();
        if (executor != null) executor.shutdown();
        try 
            if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) 
                System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
            
         catch (InterruptedException e) 
            System.out.println("Interrupted during shutdown, exiting uncleanly");
        
   

    public void run(int a_numThreads) 
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(a_numThreads));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        // now launch all the threads
        //
        executor = Executors.newFixedThreadPool(a_numThreads);

        // now create an object to consume the messages
        //
        int threadNumber = 0;
        for (final KafkaStream stream : streams) 
            executor.submit(new ConsumerTest(stream, threadNumber));
            threadNumber++;
        
    

    private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) 
        Properties props = new Properties();
        props.put("zookeeper.connect", a_zookeeper);
        props.put("group.id", a_groupId);
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");

        return new ConsumerConfig(props);
    

    public static void main(String[] args) 
        String zooKeeper = args[0];
        String groupId = args[1];
        String topic = args[2];
        int threads = Integer.parseInt(args[3]);

        ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);
        example.run(threads);

        try 
            Thread.sleep(10000);
         catch (InterruptedException ie) 

        
        example.shutdown();