Ⅰ java工程kafka传递自定义对象，消费端获取到的是null
3. 启服务
3.1 启zookeeper
启zk两种式第种使用kafka自带zk
bin/zookeeper-server-start.sh config/zookeeper.properties&
另种使用其zookeeper位于本机位于其址种情况需要修改config面sercer.properties面zookeeper址
例zookeeper.connect=10.202.4.179:2181
3.2 启 kafka
bin/kafka-server-start.sh config/server.properties
4.创建topic
bin/kafka-topics.sh --create --zookeeper 10.202.4.179:2181 --replication-factor 1 --partitions 1 --topic test
创建名testtopic副本区
通list命令查看刚刚创建topic
bin/kafka-topics.sh -list -zookeeper 10.202.4.179:2181
5.启procer并发送消息启procer
bin/kafka-console-procer.sh --broker-list localhost:9092 --topic test
启发送消息
比
test
hello boy
按Ctrl+C退发送消息
6.启consumer
bin/kafka-console-consumer.sh --zookeeper 10.202.4.179:2181 --topic test --from-beginning
启consumerconsole看procer发送消息
启两终端发送消息接受消息
都行查看zookeeper进程kafkatopic步步排查原吧
Ⅱ kafka java 通过url连接服务器
可以连接到一个网络服务器并且能够从这个服务器下载指定的URL，程序中直接使用HTTP协议。程序将定义一个输出流，下载的URL的内容将来被写入这个流，通过socket来获得输入和输出流：
view sourceprint?
01 import java.io.*;
02 import java.net.*;
03 public class HttpClient {
04 public static void main(String[] args) {
05 try {
06 // Demo参数:
07 if ((args.length != 1) && (args.length != 2))
08 throw new IllegalArgumentException("Wrong number of args");
09 //定义输出流，下载的URL内容被写入这个流
10 OutputStream to_file;
11 if (args.length == 2) to_file = new FileOutputStream(args[1]);
12 else to_file = System.out;
13 //使用URL类来把用户指定的URL解析成几个部分
14 URL url = new URL(args[0]);
15 String protocol = url.getProtocol();
16 if (!protocol.equals("http")) // 检验是否满足支持的协议
17 throw new IllegalArgumentException("Must use 'http:' protocol");
18 String host = url.getHost();
19 int port = url.getPort();
20 if (port == -1) port = 80; //如果没有指定端口，用默认端口
21 String filename = url.getFile();
22 // 打开一个连接到指定主机和端口的网络socket连接
23 Socket socket = new Socket(host, port);
24 // 通过socket来获得输入和输出流
25 InputStream from_server = socket.getInputStream();
26 PrintWriter to_server = new PrintWriter(socket.getOutputStream());
27
28 // 发送HTTP GET命令给网络服务器，指定要下载的文件
29 // 使用了一个老版本非常简单的HTTP协议
30 to_server.print("GET " + filename + "\n\n");
31 to_server.flush(); // 立即发送
32 // 现在读取服务器的响应，把接收到的内容写入文件
33 byte[] buffer = new byte[4096];
34 int bytes_read;
35 while((bytes_read = from_server.read(buffer)) != -1)
36 to_file.write(buffer, 0, bytes_read);
37 // 当服务器关闭连接时，也关闭stuff
38 socket.close();
39 to_file.close();
40 }
41 catch (Exception e) { // 发布错误
42 System.err.println(e);
43 System.err.println("Usage: java HttpClient <URL> [<filename>]");
44 }
45 }
46 }
Ⅲ 如何用Java向kafka发送json数据
发送json也可以看成字符串处理
We have 2 Options as listed below
1) If we intend to send custom java objects to procer, We need to create a serializer which implements org.apache.kafka.common.serialization.Serializer and pass that Serializer class ring creation of your procer
Code Reference below
public class PayloadSerializer implements org.apache.kafka.common.serialization.Serializer {
public void configure(Map map, boolean b) {
}
public byte[] serialize(String s, Object o) {
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(baos);
oos.writeObject(o);
oos.close();
byte[] b = baos.toByteArray();
return b;
} catch (IOException e) {
return new byte[0];
}
}
public void close() {
}
}
And set the value serializer accordingly
<entry key="value.serializer"
value="com.spring.kafka.PayloadSerializer" />
2) No need to create custom serializer class. Use the existing ByteArraySerializer, but ring send follow the process
Java Object -> String (Preferrably JSON represenation instead of toString)->byteArray