我对Kafka很陌生,开始研究生产者-消费者模型。
我为kafka producer编写了一个程序,它将从mysql读取一个表,我正在命令提示符下读取这些消息。
问题是我只能看到表中最后一列的数据。
mysql表结构和数据:
+----------+---------+----------------+
| dept | empName | salary(double) |
+----------+---------+----------------+
| Engg | Fred | 2000 |
| Engg | Bob | 3000 |
| Engg | Joe | 1000 |
| Arts | Jack | 5000 |
| Commerce | Jill | 2400 |
| Arts | James | 3000 |
| Commerce | Rob | 8700 |
+----------+---------+----------------+
Kafka制作人课程:
package com.kafka.producer;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
@SuppressWarnings("deprecation")
public class KafkaMySqlProducer {
public static void main(String[] args) {
Connection con;
PreparedStatement pstmnt;
ResultSet rs;
Properties props = new Properties();
props.put("zk.connect", "localhost:2181");
props.put("serializer.class","kafka.serializer.StringEncoder");
props.put("metadata.broker.list","localhost:9092");
ProducerConfig config = new ProducerConfig(props);
Producer producer = new Producer(config);
try {
Class.forName("com.mysql.jdbc.Driver");
con = DriverManager.getConnection("jdbc:mysql://localhost:3306/mydb","root","cloudera");
pstmnt = con.prepareStatement("select * from department");
rs = pstmnt.executeQuery();
while(rs.next()) {
String name = rs.getString(1);
String department = rs.getString(2);
String salary = " " + rs.getDouble(3);
producer.send(new KeyedMessage("dbrec", name, department, salary));
}
con.close();
} catch (ClassNotFoundException e) {
e.printStackTrace();
} catch (SQLException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
}
我的消费者的输出:
[cloudera@quickstart kafka_2.11-0.10.2.0]$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dbrec --from-beginning
[2017-06-18 10:23:22,428] WARN Error while fetching metadata with correlation id 2 : {dbrec=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-06-18 10:23:22,628] WARN The following subscribed topics are not assigned to any members in the group console-consumer-33197 : [dbrec] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
2000.0
3000.0
1000.0
5000.0
2400.0
3000.0
8700.0
我只是从表的最后一列得到数据。
谁能告诉我我在这里犯了什么错误吗?
1条答案
按热度按时间nue99wik1#
的最后一个参数
KeyedMessage
具有4个参数的构造函数是实际消息。这个类不仅仅接受你所有值的列表。
具体来说,您提供的参数是
(topic, key, partKey, message)
. 你只消耗message
从topic