程序運(yùn)行截圖如下：

后端如下：

消費(fèi)者如下：

這里啟動(dòng)kafka先要運(yùn)行zookpeer

./zookeeper-server-start.sh ../config/zookeeper.properties

這里要先修改下配置文件，在config下的server.properties：

advertised.listeners=PLATNTEXT://122.51.245.141:9092

然后運(yùn)行kafka

./kafka-server-start.sh ../config/server.properties

web端代碼關(guān)鍵代碼如下：

配置相關(guān)KafkaProducerConfig.java

package com.kafkatest.kafkatest.config;import com.kafkatest.kafkatest.common.MessageEntity; import org.apache.kafka.clients.producer.ProducerConfig; 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.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.support.serializer.JsonSerializer;import java.util.HashMap; import java.util.Map;@Configuration @EnableKafka public class KafkaProducerConfig {@Value("${kafka.producer.servers}")private String servers;@Value("${kafka.producer.retries}")private int retries;@Value("${kafka.producer.batch.size}")private int batchSize;@Value("${kafka.producer.linger}")private int linger;@Value("${kafka.producer.buffer.memory}")private int bufferMemory;public Map<String, Object> producerConfigs(){Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);props.put(ProducerConfig.RETRIES_CONFIG, retries);props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);props.put(ProducerConfig.LINGER_MS_CONFIG, linger);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}public ProducerFactory<String, MessageEntity> producerFactory(){return new DefaultKafkaProducerFactory<>(producerConfigs(),new StringSerializer(),new JsonSerializer<>());}@Beanpublic KafkaTemplate<String, MessageEntity> kafkaTemplate(){return new KafkaTemplate<>(producerFactory());} }

生產(chǎn)者如下：

ProducerCallback.java

package com.kafkatest.kafkatest.producer;import com.google.gson.Gson; import com.kafkatest.kafkatest.common.MessageEntity; import org.apache.kafka.clients.producer.RecordMetadata; import org.springframework.kafka.support.SendResult; import org.springframework.util.concurrent.ListenableFutureCallback;public class ProducerCallback implements ListenableFutureCallback<SendResult<String, MessageEntity>> {private final long startTime;private final String key;private final MessageEntity message;private final Gson gson = new Gson();public ProducerCallback(long startTime, String key, MessageEntity message) {this.startTime = startTime;this.key = key;this.message = message;}@Overridepublic void onFailure(Throwable throwable) {throwable.printStackTrace();}@Overridepublic void onSuccess(SendResult<String, MessageEntity> result) {if(result == null){return;}long elapsedTime = System.currentTimeMillis() - startTime;RecordMetadata metadata = result.getRecordMetadata();if (metadata != null) {StringBuilder record = new StringBuilder();record.append("message(").append("key = ").append(key).append(",").append("message = ").append(gson.toJson(message)).append(")").append("sent to partition(").append(metadata.partition()).append(")").append("with offset(").append(metadata.offset()).append(")").append("in ").append(elapsedTime).append(" ms");}} }

SimpleProducer.java

package com.kafkatest.kafkatest.producer;import com.kafkatest.kafkatest.common.MessageEntity; import org.apache.kafka.clients.producer.ProducerRecord; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.kafka.core.KafkaOperations; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture;@Component public class SimpleProducer {@Autowired@Qualifier("kafkaTemplate")private KafkaTemplate<String, MessageEntity> kafkaTemplate;public void send(String topic, MessageEntity message){kafkaTemplate.send(topic, message);}public void send(String topic, String key, MessageEntity entity){ProducerRecord<String, MessageEntity> record = new ProducerRecord<>(topic,key,entity);long startTime = System.currentTimeMillis();ListenableFuture<SendResult<String, MessageEntity>> future = kafkaTemplate.send(record);future.addCallback(new ProducerCallback(startTime, key, entity));}}

源碼下載如下：

https://github.com/fengfanchen/Java/tree/master/kafka_web_producer

總結(jié)

以上是生活随笔為你收集整理的Kafka笔记-kafka外网搭建及构建生产者的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。