kafka 创建topic 并使用java/.net 连接

Posted 2022-03-27 wenwenxiaoxing

创建 topic javatest

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic javatest --partitions 3 --replication-factor 1

java 连接：https://www.cnblogs.com/duanjt/p/10132116.html

pom.xml

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.3.0</version>
</dependency>

特别注意的需要设置：advertised.listeners

 

然后附上连接的 Producer 和Consumer 的代码

import java.util.Properties;
import java.util.Random;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
public class Producer 

    public static String topic = "javatest";//定义主题

    public static void main(String[] args) throws InterruptedException 
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "www.top1novel.com:9092");//kafka地址，多个地址用逗号分割
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String,String>(p);

        try 
            while (true) 
                String msg = "Hello," + new Random().nextInt(100);
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);
                kafkaProducer.send(record);
                System.out.println("消息发送成功:" + msg);
                Thread.sleep(500);
            
         finally 
            kafkaProducer.close();
        

    

　　

import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
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 Consumer 

    public static void main(String[] args) 
        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "www.top1novel.com:9092");
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
        kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));// 订阅消息

        while (true) 
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) 
                System.out.println(String.format("topic:%s,offset:%d,消息:%s", //
                        record.topic(), record.offset(), record.value()));
            
        
    

　　

使用kafka-net 也可以连接。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using KafkaNet;
using KafkaNet.Model;
using KafkaNet.Protocol;

namespace ConsoleApp1

    class Program
    
        static void Main(string[] args)
        
            Console.WriteLine("this ");


            var options = new KafkaOptions(new Uri("http://www.top1novel.com:9092"));
            
            var router = new BrokerRouter(options);
           
            var producer = new Producer(router);
            List<Message> msgArr = new List<Message>();
            msgArr.Add(new Message("你好"));
            producer.SendMessageAsync("javatest", msgArr.ToArray()).Wait(3000);
            Console.ReadKey();
        
    

　　

 另外java 的全部源代码地址： https://github.com/fmyuan2000/kafka01