Ⅰ 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