抱歉的noob问题:我写信给Kafka使用akka,但我不能看到它在Kafka控制台消费者。
写入Kafka的配置:
kafka {
bootstrap.servers = "localhost:9002"
auto.offset.reset = "earliest"
}
我用akka编写了Kafka主题的代码:
class ServiceKafkaProducer(topicName: String, actorSystem: ActorSystem, configuration: Configuration) {
val bootstrapServers: String = configuration
.getString("kafka.bootstrap.servers")
.getOrElse(
throw new Exception("No config element foe kafka.bootstrap.servers")
)
val producerSettings: ProducerSettings[String, String] = ProducerSettings(
actorSystem,
new StringSerializer,
new StringSerializer
).withBootstrapServers(bootstrapServers)
val producer: KafkaProducer[String, String] = producerSettings.createKafkaProducer()
def send(logRecordStr: String): Unit = {
Logger.debug(s"Inside ServiceKafkaProducer, writing to $topicName")
Logger.debug(logRecordStr)
producer.send(
new ProducerRecord(topicName, logRecordStr)
)
}
}
def createTag(text: String, createdBy: UUID): Unit = {
Logger.debug("Inside TagEventProducer#createTag")
val tagId = UUID.randomUUID()
val event = TagCreated(tagId, text, createdBy)
println(event)
val record = createLogRecord(event)
send(record.encode)
}
日志
```[debug] - application - Inside TagEventProducer#createTag
TagCreated(d393d223-9eb6-45e3-8610-56a3f65c84cc,scala,f5b61ca0-0ccc-4064-94c1-cba2a5a4087b)
[debug] - application - Inside ServiceKafkaProducer, writing to tags
[debug] - application - {"id":"ed27f0d1-6b6c-469b-af97-1929dc6a5cc7","action":"tag-created","data":{"id":"d393d223-9eb6-45e3-8610-56a3f65c84cc","text":"scala","createdBy":"f5b61ca0-0ccc-4064-94c1-cba2a5a4087b"},"timestamp":1542776716868}```
(编辑)我使用spotify docker图像运行Kafka,如下所示:
version: '3.5'
services:
kafka:
image: 'spotify/kafka'
hostname: kafka
environment:
- ADVERTISED_HOST=kafka
- ADVERTISED_PORT=9092
ports:
- "9092:9092"
- "2181:2181"
volumes:
- /var/run/docker.sock:/var/run/docker.sock
networks:
- kafka_net
kafkaManager:
image: 'sheepkiller/kafka-manager'
environment:
- ZK_HOSTS=kafka:2181
- APPLICATION_SECRET=letmein
ports:
- "8000:8000"
networks:
- kafka_net
networks:
kafka_net:
name: my_network
3条答案
按热度按时间qacovj5a1#
你确定你写Kafka的配置如下?
我想应该是的
localhost:9092
. 你可能在这里打错了字。to94eoyn2#
套
kafka.bootstrap.servers
至{kafka_docker_host_api}:9092
. 如果你的问题无法解决,试着用producer.flush()
. Kafka制作人不会立即发送消息。所以如果你想强制发送消息,你需要刷新。6yjfywim3#
首先,您在producer中使用了错误的端口(9002而不是9092)。尝试使用
bootstrap.servers = kafka:9092
而不是localhost:9002
因为你的广告主持人被设置为Kafka