kafka环境配置以及Java代码测试

迈不过友情╰ 2024-04-01 16:47 32阅读 0赞

[Apache Kafka][]；kafka下载地址，一定不要下源码，不要下错，要下Binary downloads

kafka3.0之后内置zk所以不需要额外下载

#### 第一步：配置文件修改 ####

**server.properties**

# broker.id 要保持唯一，只启动一台的话则不需修改
    # The id of the broker. This must be set to a unique integer for each broker.
    broker.id=0
    # 修改日志存放位置
    # A comma separated list of directories under which to store log files
    log.dirs=D://kafka//kafka_2.13-3.1.0//kafka-logs
    # 修改zk连接地址，默认本机启动则不需要修改
    # Zookeeper connection string (see zookeeper docs for details).
    # This is a comma separated host:port pairs, each corresponding to a zk
    # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
    # You can also append an optional chroot string to the urls to specify the
    # root directory for all kafka znodes.
    zookeeper.connect=localhost:2181

如果出现问题，还要添加一个参数：hostname:服务器的ip,(必须用IP连接）

**zookeeper.properties**

# 修改日志存放位置
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.
    # the directory where the snapshot is stored.
    dataDir=D://kafka//kafka_2.13-3.1.0//zookeeper-logs

#### 第二步：启动 ####

要先启动zk  
启动语句

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

然后启动kafka  
启动语句

.\bin\windows\kafka-server-start.bat .\config\server.properties

**进入/bin/windows 目录下通过下面的命令创建topic**

kafka-topics.bat --bootstrap-server localhost:9092 --create --replication-factor 3 --partitions 1 --topic test

**查看当前有哪些topic**

kafka-topics.bat --bootstrap-server localhost:9092 --list

**查看某个topic的详细信息**

kafka-topics.bat --bootstrap-server localhost:9092 --describe --topic test

**修改topic的配置**

kafka-topics.bat --bootstrap-server localhost:9092 --alter --topic test --partitions 3

删除topic会报错，此时停掉所有的cmd，然后将日志文件删除，重新启动可以解决

**启动生产者命令/bin/window 目录下**

kafka-console-producer.bat --broker-list localhost:9092 --topic test

**启动消费者命令/bin/window 目录下**

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

然后就可以快乐的发送啦！

Java代码：

maven添加配置

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.12</artifactId>
                <version>2.1.1</version>
                <scope>provided</scope>
            </dependency>
            
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.6.3</version>
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-streams</artifactId>
                <version>1.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
                <version>3.12.0</version>
            </dependency>

生产者代码：

import java.util.Properties;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    
    public class KafkaProducerTest implements Runnable {
    
        private final KafkaProducer<String, String> producer;
        private final String topic;
        private String clientid;
        public KafkaProducerTest(String topicName,String clientid) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.0.108:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("key.serializer", StringSerializer.class.getName());
            props.put("value.serializer", StringSerializer.class.getName());
            this.producer = new KafkaProducer<String, String>(props);
            this.topic = topicName;
            this.clientid = clientid;
        }
    
        @Override
        public void run() {
            int messageNo = 1;
            try {
                for(;;) {
                    String messageStr= "你好，这是第"+messageNo+"条数据 clientid=" + clientid;
                    producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                    //生产了100条就打印
                    if(messageNo%100==0){
                        System.out.println("发送的信息:" + messageStr);
                    }
                    //生产1000条就退出
                    if(messageNo == 1000){
                        System.out.println("成功发送了"+messageNo+"条");
                        break;
                    }
                    messageNo++;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                producer.close();
            }
        }
    
        public static void main(String args[]) {
            KafkaProducerTest test1 = new KafkaProducerTest("test", "clientid1");
            Thread thread1 = new Thread(test1);
            thread1.start();
        }
    }

消费者代码：

import java.util.Arrays;
    import java.util.Properties;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    
    public class KafkaConsumerTest implements Runnable {
    
        private final KafkaConsumer<String, String> consumer;
        private ConsumerRecords<String, String> msgList;
        private final String topic;
        private String clientid;
        private static final String GROUPID = "groupA";
    
        public KafkaConsumerTest(String topicName,String clientid) {
    
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.0.108:9092");
            props.put("group.id", GROUPID);
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("auto.offset.reset", "earliest");
            props.put("key.deserializer", StringDeserializer.class.getName());
            props.put("value.deserializer", StringDeserializer.class.getName());
            this.consumer = new KafkaConsumer<String, String>(props);
            this.topic = topicName;
            this.consumer.subscribe(Arrays.asList(topic));
            this.clientid = clientid;
        }
    
        @Override
        public void run() {
            int messageNo = 1;
            System.out.println("---------开始消费---------");
            try {
                for (;;) {
                    msgList = consumer.poll(1000);
                    if(null!=msgList&&msgList.count()>0){
                        for (ConsumerRecord<String, String> record : msgList) {
                            //消费100条就打印 ,但打印的数据不一定是这个规律的
                            if(messageNo%100==0){
                                System.out.println(messageNo+"=======成功消费：receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
                            }
                            //当消费了1000条就退出
                            if(messageNo == 1000){
                                break;
                            }
                            messageNo++;
                        }
                    }else{
                        Thread.sleep(1000);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                consumer.close();
            }
        }
    
        public static void main(String args[]) {
            KafkaConsumerTest test1 = new KafkaConsumerTest("test", "clientid1");
            Thread thread1 = new Thread(test1);
            thread1.start();
        }
    }

自测没问题！

[Apache Kafka]: https://kafka.apache.org/downloads.html