java连接外网kafka实现producer和consumer（高级API）

今天药忘吃喽~ 2022-02-24 11:14 219阅读 0赞

踩坑踩过来的，记录一下。。。

#### 背景 ####

在华为云上搭建的kafka的为分布式环境，配置文件server.properties是一个基础配置，[云服务器搭建kafka伪分布式环境][kafka]  
没有配置外网连接的一些属性，先来配置一下：

需要在原来的配置基础下加入如下配置：

listeners=PLAINTEXT://0.0.0.0:8180        # 端口可以自己设定
    advertised.host.name=hwyuntrx             # 云服务器主机名   自己设定
    advertised.port=8180                      # 端口

配置好了之后启动kafka，**注意：你所设定的端口，记得在云服务器的控制台配置安全组开放，否则是连不上的。**

**我已经在kafka上创建了一个topic：叫 test1**

### 代码实现 ###

###### 先导包 ######

<dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka-clients</artifactId>
          <version>0.11.0.0</version>
        </dependency>
        
        <dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka_2.11</artifactId>
          <version>0.11.0.0</version>
        </dependency>

#### producer 生产者代码 ####

package test.kafka;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * kafka producer */
    public class ProducerDemo { 
    
        public static void main(String[] args) { 
    
            // 配置信息
            Properties props = new Properties();
            // kafka 集群 
            //props.put("bootstrap.servers", "hwyuntrx:9092");
            props.put("bootstrap.servers", "hwyuntrx:8180");
            // 应答级别
            props.put("acks", "all");
            // 重试此时
            props.put("retries", 0);
            // 批量大小
            props.put("batch.size", 16384);
            // 提交延时
            props.put("linger.ms", 1);
            // 缓存
            props.put("buffer.memory", 33554432);
            // KV的序列化类
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 创建生产者对象
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            // 发送十条消息 
            for(int i = 0; i < 10; i++)
            { 
                producer.send(new ProducerRecord<String, String>("test1", "消息--"+i));//指定向test1 发送消息
                try { 
                    Thread.sleep(2000);//停顿两秒 便于观察
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    		// 最后记得关闭连接 
            producer.close();
        }
    }

#### consumer 消费者代码 ####

package test.kafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * @Author:renxin.tang * @Desc: * @Date: Created in 19:52 2019/4/3 */
    public class ConsumerDemo { 
        public static void main(String[] args) { 
            Properties props = new Properties();
            props.put("bootstrap.servers", "hwyuntrx:8180");
            props.put("group.id", "tests");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList("test1"));// 设置我创建的topic：test1 同时可以订阅多个topic
            while (true) { 
                ConsumerRecords<String, String> records = consumer.poll(10);
                for (ConsumerRecord<String, String> record : records)
                { 
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
                    System.out.println();
                }
            }
        }
    
    }

### 测试 ###

1、开启zookeeper集群  
2、开启kafka  
3、运行consumer 代码  
4、运行producer代码

运行结果：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70]

### 补充—java创建topic ###

public static void main(String[] args) { 
            // 创建topic
            Properties props = new Properties();
            props.put("bootstrap.servers", "hwyuntrx:8180");
            AdminClient adminClient = AdminClient.create(props);
            ArrayList<NewTopic> topics = new ArrayList<NewTopic>();
            NewTopic newTopic = new NewTopic("topic-trx", 1, (short) 1);// 创建 topic ： topic-trx
            topics.add(newTopic);
            CreateTopicsResult result = adminClient.createTopics(topics);
            try { 
                result.all().get();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }

运行结果：

![在这里插入图片描述][20190403212652685.png]  
done！

[kafka]: https://blog.csdn.net/weixin_43161811/article/details/89004918
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70]: /images/20220224/cd6b334083f34b899597869ed04b5197.png
[20190403212652685.png]: /images/20220224/8c61c7280e4046c0a65997b1a2be4ded.png