2. 程式人生 > >kafka傳送自定義訊息體（物件、陣列）

kafka傳送自定義訊息體（物件、陣列）

阿新 • • 發佈：2019-02-07

在前面簡單搭建了Windows上的kafka環境，並使用命令列測試可以執行之後（環境請參考：http://blog.csdn.net/u014104286/article/details/75040932）我們會考慮怎麼使用kafka；先試著傳送一個簡單的訊息，傳送成功之後是否需要傳送自定義的訊息類尼？怎麼傳送自定義的訊息類，如果我要傳送一個集合呢？下面我們來一一解決我們的問題。

準備工作：

1.需要搭建並測試成功的kafka環境，並啟動zookeeper和kafka服務。

2.建立一個可用的maven專案

3.新增開發kafkaka的依賴：

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka_2.11 -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.2.0</version>
</dependency>
準備工作完成。

1.首先我們要傳送第一個訊息，訊息型別為String：

Producer傳送訊息類：

public class SimpleProducer {
    public static void main(String[] args) throws Exception{         
          //Assign topicName to string variable
          String topicName = "newtest001";
          // create instance for properties to access producer configs  
          Properties props = new Properties();
          //Assign localhost id
          props.put("bootstrap.servers", "localhost:9092");
          //Set acknowledgements for producer requests.     
          props.put("acks", "all");
          //If the request fails, the producer can automatically retry,
          props.put("retries", 0);
          //Specify buffer size in config
          props.put("batch.size", 16384);
          //Reduce the no of requests less than 0  
          props.put("linger.ms", 1);
          //The buffer.memory controls the total amount of memory available to the producer for buffering.  
          props.put("buffer.memory", 33554432);
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
          Producer<String, String> producer = new KafkaProducer<String, String>(props);
 
          for(int i = 0; i < 10; i++)
             producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
                   System.out.println("Message sent successfully");
                   producer.close();
       }
}

Consumer接收訊息類：
public class SimpleConsumer {
    public static void main(String[] args) throws Exception {
          //Kafka consumer configuration settings
          String topicName = "newtest001";
          Properties props = new Properties();
 
          props.put("bootstrap.servers", "localhost:9092");
          props.put("group.id", "test");
          props.put("enable.auto.commit", "true");
          props.put("auto.commit.interval.ms", "1000");
          props.put("session.timeout.ms", "30000");
          props.put("key.deserializer",
             "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("value.deserializer",
             "org.apache.kafka.common.serialization.StringDeserializer");
          @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
 
          //Kafka Consumer subscribes list of topics here.
          consumer.subscribe(Arrays.asList(topicName));
 
          //print the topic name
          System.out.println("Subscribed to topic "+ topicName);
          while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records)
 
             // print the offset,key and value for the consumer records.
             System.out.printf("offset = %d, key = %s, value = %s\n",
                record.offset(), record.key(), record.value());
          }
 
       }
}
以上內容參考：https://www.w3cschool.cn/apache_kafka/apache_kafka_simple_producer_example.html

啟動Consumers類，等待並接收Producer傳送的訊息：執行Producer傳送訊息，Consumers接收到的訊息：


說明能成功傳送和接收訊息。上面傳送的訊息都是字元，我們如果需要傳送一個PerSon這樣JavaBean那怎麼做呢？

我們可以先觀察上面的Producer和Consumers，他們在例項化一個Producer和一個Consumers的時候需要一些引數，其中有： props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
       props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");可見是對訊息的key和value的序列化指定類。我們到org.apache.kafka.common.serialization.StringSerializer中可以看見這個類實現了

org.apache.kafka.common.serialization.Deserializer<T>和org.apache.kafka.common.serialization.Serializer<T>


我們也分別實現序列化和反序列化的藉口即可：

DecodeingKafka類：

import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;
import com.ys.test.SpringBoot.zktest.util.BeanUtils;

public class DecodeingKafka implements Deserializer<Object> {

	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {
	}

	@Override
	public Object deserialize(String topic, byte[] data) {
		return BeanUtils.byte2Obj(data);
	}

	@Override
	public void close() {
		
	}
}

EncodeingKafka類：
import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
import com.ys.test.SpringBoot.zktest.util.BeanUtils;
public class EncodeingKafka implements Serializer<Object> {
	@Override
	public void configure(Map configs, boolean isKey) {
		
	}
	@Override
	public byte[] serialize(String topic, Object data) {
		return BeanUtils.bean2Byte(data);
	}
	/*
	 * producer呼叫close()方法是呼叫
	 */
	@Override
	public void close() {
		System.out.println("EncodeingKafka is close");
	}
}

之後我們需要定義JavaBean物件怎麼序列化和反序列化，我們使用ObjectOutputStream和ObjectInputStream實現。（大家也可以考慮更高效的序列化方法）

BeanUtils類：

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

public class BeanUtils {
	private BeanUtils() {}
	/**
	 * 物件序列化為byte陣列
	 * 
	 * @param obj
	 * @return
	 */
	public static byte[] bean2Byte(Object obj) {
		byte[] bb = null;
		try (ByteArrayOutputStream byteArray = new ByteArrayOutputStream();
			 ObjectOutputStream outputStream = new ObjectOutputStream(byteArray)){
			outputStream.writeObject(obj);
			outputStream.flush();
			bb = byteArray.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return bb;
	}
	/**
	 * 位元組陣列轉為Object物件
	 * 
	 * @param bytes
	 * @return
	 */
	public static Object byte2Obj(byte[] bytes) {
		Object readObject = null;
		try (ByteArrayInputStream in = new ByteArrayInputStream(bytes);
			 ObjectInputStream inputStream = new ObjectInputStream(in)){
			 readObject = inputStream.readObject();
		} catch (Exception e) {
			e.printStackTrace();
		} 
		return readObject;
	}
}

PerSon.java：

public class PerSon implements Serializable{
    /**
     *
     */
    private static final long serialVersionUID = 1L;
    private long userid;
    private String name;
    private int age;
    private String addr;
    private String eMail;
    private String userRole;
    private IDCard card;
set... get...

SimpleProducerPerSon.java：訊息生產者： 
import java.util.Arrays;
import java.util.List;
//import util.properties packages
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;
//import simple producer packages
import org.apache.kafka.clients.producer.Producer;
//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import com.ys.test.SpringBoot.model.IDCard;
import com.ys.test.SpringBoot.model.PerSon;
public class SimpleProducerPerson {
	public static void main(String[] args) throws Exception{
	      
	      //Assign topicName to string variable
	      String topicName = "mypartition001";
	      // create instance for properties to access producer configs   
	      Properties props = new Properties();
	      //Assign localhost id
	      props.put("bootstrap.servers", "localhost:9092");
	      //Set acknowledgements for producer requests.      
	      props.put("acks", "all");
	      //If the request fails, the producer can automatically retry,
	      props.put("retries", 0);
	      props.put("metadata.fetch.timeout.ms", 30000);
	      //contorller the send method ：sync or async default ： sync
	      //Specify buffer size in config
	      props.put("batch.size", 16384);
	      //Reduce the no of requests less than 0   
	      props.put("linger.ms", 1);
	      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
	      props.put("buffer.memory", 33554432);
	      props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	      props.put("value.serializer", "com.ys.test.SpringBoot.zktest.encode.EncodeingKafka");
//	      props.put("partitioner.class", "繼承了Partition的類，實現的是根據指定的演算法把訊息推送到指定的分割槽中com.ys.test.SpringBoot.zktest.util.MyPartition");
	      
	      Producer<String, Object> producer = new KafkaProducer<String, Object>(props);
		long startTimes = System.currentTimeMillis();
	      System.out.println();
	      
	      for(int i = 0; i < 2; i++){
	    	  
	    	  final int index = i;
	    	  PerSon perSon = new PerSon();
	    	  perSon.setAge(i);
	    	  perSon.setAddr("My Producer TEST001"+i);
              perSon.setName("MyTest "+i);
              IDCard card = new IDCard();
              card.setCardName("MyTest"+i+"'s idCard");
	    	  card.setCardid(10000000000L);
	    	  perSon.setCard(card);
	    	  
	    	  List<PerSon> asList = Arrays.asList(perSon,perSon);
//	    	  producer.send(new ProducerRecord<String, Object>(topicName,Integer.toString(i),asList));
//	          producer.send(new ProducerRecord<String, Object>(topicName, Integer.toString(i), perSon));
	          producer.send(new ProducerRecord<String, Object>(topicName, Integer.toString(i), asList), new Callback() {
				
				@Override
				public void onCompletion(RecordMetadata metadata, Exception exception) {
					if (metadata != null) {
						System.out.println(index+"  傳送成功："+"checksum: "+metadata.checksum()+" offset: "+metadata.offset()+" partition: "+metadata.partition()+" topic: "+metadata.topic());
					}
					if (exception != null) {
						System.out.println(index+"異常："+exception.getMessage());
					}
				}
			});
	      }
	      producer.close();
	   }
}

SimpleConsumersPerSon.java 訊息接收者： 
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class SimpleConsumerPerSon {
	public static void main(String[] args) throws Exception {

	      String topicName = "mypartition001";
	      Properties props = new Properties();
	      
	      props.put("bootstrap.servers", "localhost:9092");
	      props.put("group.id", "partitiontest05");
	      props.put("enable.auto.commit", "true"); 
	      props.put("auto.commit.interval.ms", "1000");
	      props.put("session.timeout.ms", "30000");
	      
	      //要傳送自定義物件，需要指定物件的反序列化類
	      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	      props.put("value.deserializer", "com.ys.test.SpringBoot.zktest.encode.DecodeingKafka");
	      
	      //使用String時可以使用系統的反序列化類
//	      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//	 	  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	      @SuppressWarnings("resource")
		KafkaConsumer<String, Object> consumer = new KafkaConsumer<String, Object>(props);
	      //Kafka Consumer subscribes list of topics here.
	      consumer.subscribe(Arrays.asList(topicName));
	      //print the topic name
	      System.out.println("Subscribed to topic "+ topicName);
	      
	      
	      while (true) {
	    	  ConsumerRecords<String, Object> records = consumer.poll(100);
	         for (ConsumerRecord<String, Object> record : records)
	         // print the offset,key and value for the consumer records.
//	         System.out.printf("offset = %d, key = %s, value = %s\n", 
//	            record.offset(), record.key(), record.value().toString());
	        	 
	        	 System.out.println(record.toString());
	      }
	      
	   }
}

以上的傳送者和接收者他們的key的序列化類還是StringDeserializer，但是value的序列化需要指定為我們自己的。

執行接收者和傳送者，觀察結果：（結果是傳送一個集合和傳送一個物件）

傳送一個物件person接收的結果：

ConsumerRecord(topic = mypartition001, partition = 0, offset = 29, CreateTime = 1502457680160, checksum = 3691507046, serialized key size = 1, serialized value size = 391, key = 0, value = PerSon [userid=0, name=MyTest 0, age=0, addr=My Producer TEST0010, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest0's idCard]])
ConsumerRecord(topic = mypartition001, partition = 0, offset = 30, CreateTime = 1502457680175, checksum = 1443537499, serialized key size = 1, serialized value size = 391, key = 1, value = PerSon [userid=0, name=MyTest 1, age=1, addr=My Producer TEST0011, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest1's idCard]])

傳送asList集合的結果：
ConsumerRecord(topic = mypartition001, partition = 0, offset = 31, CreateTime = 1502457689533, checksum = 3469353517, serialized key size = 1, serialized value size = 524, key = 0, value = [PerSon [userid=0, name=MyTest 0, age=0, addr=My Producer TEST0010, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest0's idCard]], PerSon [userid=0, name=MyTest 0, age=0, addr=My Producer TEST0010, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest0's idCard]]])
ConsumerRecord(topic = mypartition001, partition = 0, offset = 32, CreateTime = 1502457689552, checksum = 1930168239, serialized key size = 1, serialized value size = 524, key = 1, value = [PerSon [userid=0, name=MyTest 1, age=1, addr=My Producer TEST0011, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest1's idCard]], PerSon [userid=0, name=MyTest 1, age=1, addr=My Producer TEST0011, eMail=null, userRole=null, card=IDCard [cardid=10000000000, cardName=MyTest1's idCard]]])

這樣我們不管是傳送的是一個物件還是一個集合我們都可以正確傳送和接收了。

相關推薦

kafka傳送自定義訊息體（物件、陣列）

在前面簡單搭建了Windows上的kafka環境，並使用命令列測試可以執行之後（環境請參考：http://blog.csdn.net/u014104286/article/details/75040932）我們會考慮怎麼使用kafka；先試著傳送一個簡單的訊息，傳送成功之後

自定義Button形狀（圓形、橢圓） shape

這裡介紹2種改變Button形狀的方法，一個是直接通過設定一張圓形背景圖片， 另一種方法是通過引用自定義的shape 檔案。具體參見如下：   --------main_menu_btnshape.xml              自定義形狀xml檔案 <?xml version="1.0" e

WPF 自定義滾動條（ScrollView、ScrollBar）樣式

原文: WPF 自定義滾動條（ScrollView、ScrollBar）樣式 一、滾動條基本樣式 本次修改Scrollview及ScrollBar滾動條樣式是通過純樣式實現的。修改的內容包含滾動條的顏色，上下按鈕的隱藏。另外添加了滑鼠經過滾動條動畫。 style樣式如下： <!-- Scr

vue 響應式修改資料（物件、陣列）及難點（坑）

背景 1. Vue不能檢測物件屬性的新增、刪除 2. Vue不能檢測陣列項的賦值修改（根據index）、length的修改   物件  1. $set、set、$delete、delete // $set是Vue.set的別名 this.$set(this

接入微信公眾平臺開發之使用者關注（取消）事件觸發後臺自定義訊息體通知給使用者的實現過程

1.需求：使用者關注公眾號後回覆給使用者一個字串，字串不能重複使用即如果a使用者關注公眾號後商戶後臺回覆給使用者字串str1後，b使用者關注就是其他字串，且a使用者取消關注再次關注不回覆訊息體 2.實現過程： 　　①首先配置伺服器url並開啟，再次過程中需要微信後臺與商戶後臺進行通訊，所以，微信後臺會發送

MFC中傳送自定義訊息機制，PostMessage和SendMessage方式

       MFC中有種訊息佇列，使用PostMessage()或者SendMessage()給窗體傳送自定義訊息，當窗體接收到訊息時呼叫該訊息對應繫結的方法。 PostMessage:把訊息投放到執行緒的訊息佇列，不等訊息被處理就立即返回；Send

vector 的輸入輸出和結構體自定義排序模板（蒟蒻版）

#include<bits/stdc++.h> using namespace std; struct node { int id; }b[10]; vector<node>a; bool cmp(const node a,const node b) { retu

Pythpn整合itchat定時傳送自定義訊息到微信群

完整程式碼如下： import itchat import time from apscheduler.schedulers.blocking import BlockingScheduler # 登陸，enableCmdQR 值為true 或 負數 為負數的原因有的手機無法識別二維碼，

ROS---傳送自定義訊息，接收訊息

上一篇已經建好了hello包及其內部的檔案，本篇講訊息 接收訊息 hello_node.cpp 作一些修改 #include "iostream" #include "../include/hello/add.h" #include "ros/ros.h

Netty 伺服器客戶端，傳送自定義訊息

Netty 5 的包 伺服器，四個Java檔案 1            SubReqServer.java [java] view plain copy  print? package com.t

FastReport調用Delphi中的自定義函數（人民幣大寫金額）mtm

分享圖片 人民幣 IT report text 定義 eat fas func 1. 在 FormCreate 中向FastReprot添加函數 (fPrint)窗口 procedure TfPrint.FormCreate(Sender: TObject); fr

簡單的自定義標題欄（不使用Toolbar）

比較簡單的自定義標題欄，這裡直接封裝成一個類似控制元件的樣子，先上效果圖： 我們可以先寫一個BaseView來，用來做標題欄的基礎佈局： /** * 自定義View的基類 * @author */ public class BaseView extends FrameL

OpenLayers官方示例詳解九自定義多邊形樣式（Custom Polygon Styles）

目錄 一、示例簡介 二、程式碼詳解 一、示例簡介     這個示例演示怎樣為多邊形要素建立自定義的樣式。     在這個示例中，將為多邊形建立兩種不同的樣式： 第一個樣式是多邊形的整體樣式 第二個樣式是為多邊形的各個頂點

自定義view流程（結合原始碼分析）

一、View的繪製流程 主要是：測量（measure）、佈局（layout）、繪製（draw）三大流程。 對於一個普通View（不是容器） 主要是關心測量和繪製兩個過程，測量可以確定自身的寬、高、大小，繪製可以顯示出view的具體內容（呈現在螢幕上的）。 對於

富文字編輯器UEditor自定義工具欄（一、基礎配置與字型、背景色、行間距、超連結實現）

導讀：UEditor 是由百度「FEX前端研發團隊」開發的所見即所得富文字web編輯器，功能強大，可定製，是一款優秀的國產線上富文字編輯器，編輯器內可插入圖片、音訊、視訊等。 一、UEditor自定義工具欄效果圖如下： 二、UEditor富文字編輯器環境搭建及專案引用 環境搭建不再贅述，請自行查閱或者參考以

UITabBar自定義方式三（中間有按鈕）

1.MainTableViewController #import "MainTabBarViewController.h" #import "IWHomeViewController.h" #import "IWMessageViewController.h" #i

Win10自定義DNS對映（修改hosts檔案）

hosts檔案位置：C:\Windows\System32\drivers\etchosts檔案作用：將web瀏覽器中輸入的域名對映成為IP地址作用：通過修改我們就可以將自定義網址對映到本機或者任意一個IP而在Win10下是無法修改hosts檔案的想要達到目的需要費一點事：1

06.自定義s.repalce（old,new,times）

def str_replace(s,old,new,*times): #未寫len(times)>2的情況，可以自行完善 len_old=len(old) len_s=len(s) result='' if not times or s.count(ol

Android簡單自定義進度條（小白進）

今天，我在騰訊課堂學習了自定義一個進度條，視訊連結：https://ke.qq.com/course/144239#term_id=100163342，我根據此視訊教程，寫了關於此程式碼的詳細註釋，非常簡單，非常適合小白學習。 主要步驟是: 1.

自定義控制元件（陰影，漸變）

一、陰影： Paint 類定義了一個名為 setShadowLayer 的方法： publicvoid setShadowLayer(float radius,float dx,float dy,int shadowColor)， 引數意義如下： radius：陰影半徑 d