kafka常见问题

1 启动advertised.listeners配置异常:

java.lang.IllegalArgumentException: requirement failed: advertised.listeners cannot use the nonroutable meta-address 0.0.0.0. Use a routable IP address.
    at scala.Predef$.require(Predef.scala:277)
    at kafka.server.KafkaConfig.validateValues(KafkaConfig.scala:1203)
    at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1170)
    at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:881)
    at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:878)
    at kafka.server.KafkaServerStartable$.fromProps(KafkaServerStartable.scala:28)
    at kafka.Kafka$.main(Kafka.scala:82)
    at kafka.Kafka.main(Kafka.scala)

1.1 解决方法:修改server.properties

advertised.listeners=PLAINTEXT://{ip}:9092  # ip可以内网、外网ip、127.0.0.1 或域名

1.2 解析:

server.properties中有两个listeners。 listeners:启动kafka服务监听的ip和端口,可以监听内网ip和0.0.0.0(不能为外网ip),默认为java.net.InetAddress.getCanonicalHostName()获取的ip。advertised.listeners:生产者和消费者连接的地址,kafka会把该地址注册到zookeeper中,所以只能为除0.0.0.0之外的合法ip或域名 ,默认和listeners的配置一致。

2 启动PrintGCDateStamps异常

[0.004s][warning][gc] -Xloggc is deprecated. Will use -Xlog:gc:/data/service/kafka_2.11-0.11.0.2/bin/../logs/kafkaServer-gc.log instead.
Unrecognized VM option 'PrintGCDateStamps'
Error: Could not create the Java Virtual Machine.
Error: A fatal exception has occurred. Program will exit.

2.1 解决方法: 更换jdk1.8.x版本或者使用>=kafka1.0.x的版本。

2.2 解析:

只有在jdk1.9并且kafka版本在1.0.x之前的版本才会出现。

3 生成者发送message失败或消费者不能消费(kafka1.0.1)

#(java)org.apache.kafka警告
Connection to node 0 could not be established. Broker may not be available.


# (nodejs) kafka-node异常 (执行producer.send后的异常)
{ TimeoutError: Request timed out after 30000ms
    at new TimeoutError (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\errors\TimeoutError.js:6:9)
    at Timeout.setTimeout [as _onTimeout] (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\kafkaClient.js:737:14)
    at ontimeout (timers.js:466:11)
    at tryOnTimeout (timers.js:304:5)
    at Timer.listOnTimeout (timers.js:264:5) message: 'Request timed out after 30000ms' }

3.1 解决方法: 检查advertised.listeners的配置(如果有多个Broker可根据java版本的对应的node号检查配置),判断当前的网络是否可以连接到地址(telnet等)

4 partitions配置的值过小造成错误(kafka1.0.1)

#(java)org.apache.kafka(执行producer.send)
Exception in thread "main" org.apache.kafka.common.KafkaException: Invalid partition given with record: 1 is not in the range [0...1).
    at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:908)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:778)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:768)
    at com.wenshao.dal.TestProducer.main(TestProducer.java:36)


# (nodejs) kafka-node异常 (执行producer.send后的异常)
{ BrokerNotAvailableError: Could not find the leader
    at new BrokerNotAvailableError (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\errors\BrokerNotAvailableError.js:11:9)
    at refreshMetadata.error (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\kafkaClient.js:831:16)
    at D:\project\node\kafka-test\src\node_modules\kafka-node\lib\client.js:514:9
    at KafkaClient.wrappedFn (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\kafkaClient.js:379:14)
    at KafkaClient.Client.handleReceivedData (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\client.js:770:60)
    at Socket.<anonymous> (D:\project\node\kafka-test\src\node_modules\kafka-node\lib\kafkaClient.js:618:10)
    at Socket.emit (events.js:159:13)
    at addChunk (_stream_readable.js:265:12)
    at readableAddChunk (_stream_readable.js:252:11)
    at Socket.Readable.push (_stream_readable.js:209:10) message: 'Could not find the leader' }

4.1 解决方法: 修改num.partitions的值,partitions在是在创建topic的时候默认创建的partitions节点的个数,只对新创建的topic生效,所有尽量在项目规划时候定一个合理的值。也可以通过命令行动态扩容()

./bin/kafka-topics.sh --zookeeper  localhost:2181 --alter --partitions 2 --topic  foo

 

Apache Kafka

Kafka说明

Apache Kafka® 是一个分布式流媒体平台,这是什么意思呢?

分布式流媒体平台它有三大特性:

  • 发布与订阅流媒体数据,类似于消息队列或企业消息传递系统
  • 能够容错并持久性存储流媒体数据
  • 处理流媒体数据

Kafka通常适用于两大类应用场景:

  • 在系统或应用程序之间构建实时数据流管道,使其可靠地获取数据
  • 在应用程序中对数据流进行实时转换或响应

要了解Kafka如何做这些事情,让我们深入探讨Kafka的能力。

首先是几个概念:

  • Kafka作为一个集群运行在一个或多个可跨多个数据中心的服务器上。
  • Kafka集群以称为topics类别存储记录流。
  • 每条记录流都由一个键、值和时间戳组成。

Kafka有四个核心API:

  • 生产者(Producer API )允许应用程序发布一条数据记录到一个或更多的Kafka topics。
  • 消费者(Consumer API)允许应用程序订阅一个或多个主题,并处理为其生成的记录流。
  • 数据流(Streams API)允许应用程序充当流处理器( stream processor),使用来自一个或多个topics输入流,并将输出流生成到一个或多个输出topics,从而有效地将输入流转换为输出流。
  • 连接器(Connector API)允许构建并运行可重用的生产者或消费者,将Kafka topics连接到现有的应用程序或数据系统。例如,数据库的连接器可以捕获每个表的更改。

1556898600-9189-kafka-apis

在Kafka中,客户机和服务器之间的通信是通过一种简单、高性能、与语言无关的TCP协议来完成。 此协议已经版本化,并保持与旧版本的向后兼容性。我们为Kafka提供Java客户端,客户端可以使用多语言版本。

Topics and Logs

让我们首先深入探讨Kafka为记录流提供的核心抽象 – topic。

topic是发布记录的类别或源名称。Kafka中的Topics总是多个订阅用户;也就是说,一个topic可以有0个、1个或多个订户订阅者。

对于每个topic,Kafka群集都维护一个分区日志,如下所示:

 

1556898598-1342-log-anatomy每个分区都是一个有序的,不可变的记录序列,不断附加到一个结构化的日志中。分区中的记录每个都被分配一个 offset 的顺序ID号,它标识分区中的每个记录。

Kafka集群持久地保留所有已发布的记录,无论它们是否已被消耗,可以使用可配置方式设置的过期时间。例如,如果保留策略设置为两天,则在发布记录后的两天内,它可供使用,之后将被丢弃以释放空间。Kafka的性能在数据大小方面实际上是恒定的,因此长时间存储数据不是问题。

1556898607-4067-log-consumer事实上,在消费者的日志中使用偏移量(offset)或分区来保持唯一的元数据。这种偏移(offset)由消费者控制:通常消费者在读取记录时会线性地提高其偏移量(offset),但事实上,由于消费者控制偏移量(offset)的位置,它可以按照自己喜欢的任何顺序消费记录。例如,消费者可以重置为已处理过或较旧的偏移量(offset)以重新处理过去的数据,或者跳到最近的记录并从“现在”开始消费。

这些功能组合意味着Kafka消费者非常简单,他们可以来来往往对集群或其他消费者没有太大影响。例如,您可以使用我们的命令行工具“tail”任何主题的内容,而无需更改任何现有消费者所消费的内容。

日志中的分区有多种用途。首先,它们允许日志扩展到超出单个服务器的大小。每个单独的分区必须适合托管它的服务器,但topic可能有许多分区,因此它可以处理任意数量的数据。其次,在一点上他们更像是并行的单元。

分配

日志的分区分布在Kafka集群中的服务器上,每个服务器处理数据并请求分区的共享。每个分区都在可配置数量的服务器上进行复制,以实现容错。

每个分区都有一个服务器充当“领导者(leader)”,0个或多个服务器充当“追随者(followers)”。领导者处理分区的所有读取和写入请求,而关注者被动地复制领导者。如果领导者出现故障,其中一个追随者将自动成为新的领导者。每个服务器都充当其某些分区的领导者和其他服务器的追随者,因此负载在群集中很均衡。

地理复制

Kafka MirrorMaker为群集提供地理复制的支持。使用MirrorMaker,消息跨多个数据中心或云区域进行复制。你可以使用它在active/passive方案中进行备份和恢复; 或者在active/active方案中,按地理的方式,使数据更接近用户,或支持数据位置要求。

生产者

生产者将数据发布到他们选择的topics。生产者负责选择要分配给topic中哪个分区的记录。这可以通过循环方式完成,只是为了平衡负载,或者可以根据一些语义分区功能(例如:基于记录中的某些键)来完成。

消费者

消费者使用 consumer group 名称标记自己,每一个记录都会发布到一个topic中,并传递给每一个订阅的 consumer group 中其中一个消费者。消费者实例可以在同一个进程中,也可以在不同的机器

如果所有消费者实例具有相同的 consumer group,那么记录将在消费者实例上进行负载平衡(只有其中一个能收到消息)。

如果所有消费者实例具有不同的 consumer groups,那么每个记录将使用广播的方式,发送到所有 consumer group 进程。

 

1556898599-7054-consumer-groups两个服务器Kafka群集,托管四个分区(P0-P3),包含两个 consumer groups。 group A有两个消费者, group B有四个消费者。

然而,更常见的是,我们发现topics具有少量的 consumer groups,每个“logical subscriber”一个。每个组由许多用于可伸缩性和容错的消费者实例组成。这只不过是发布 – 订阅语义,其中订阅者是消费者群集而不是单个进程。

在Kafka中实现消费的方式是通过在消费者实例上划分日志中的分区,以便每个实例在任何时间点都是分配的“公平份额”的独占消费者。这个维护组成员身份的过程是由kafka协议动态处理的。如果有新的消费者实例加入该组,他们将从该组的其他成员接管一些分区; 如果实例出现故障,那么其分区将分配给其余消费者实例。

Kafka仅提供分区内记录的总订单,而不是主题中不同分区之间的记录。对于大多数应用程序而言,按分区排序与按键分区数据的能力相结合就足够了。但是,如果需要对记录进行总排序,可以使用只有一个分区的主题来实现,但这将意味着每个 consumer group 只有一个消费者进程。

多租户

您可以将Kafka部署为多租户解决方案。通过配置哪些主题可以生成或使用数据来启用多租户。配额也有运营支持。管理员可以定义和强制执行配额,以控制客户端使用的代理资源。有关更多信息,请参阅安全文档

担保

在高级别Kafka提供以下保证:

  • 生产者按顺序将消息发送到特定主题分区。也就是说,如果记录数据M1由与数据M2是由同一个生产者发送,并且首先发送M1,则M1将具有比M2更低的偏移并在日志中更早出现。
  • 消费者实例按照它们存储在日志中的顺序查看记录。
  • 对于具有复制因子N的主题,我们将容忍最多N-1个服务器故障,而不会丢失任何提交到日志的记录。

下一篇,将Kafka集成到Spring中。

有关Kafka提供的API和功能的更多信息,请参阅官方文档

Kafka与Zookeeper常用操作

一、Kafka操作

1.启动kafka命令:

#cd /opt/kafka_2.10-0.10.1.1/bin;

# ./kafka-server-start.sh /opt/kafka_2.10-0.10.1.1/config/server.properties &;

2.停止kafka命令:

# ./kafka-server-stop.sh

3.创建Topic:(创建一个名为test的topic,只有一个副本,一个分区。)

#./kafka-topics.sh –create –zookeeper 127.0.0.1:2181 –replication-factor 1 –partitions 1 –topic test

4.列出所有Topic:

#./kafka-topics.sh -list -zookeeper 127.0.0.1:2181

5.启动Producer并发送消息:

#./kafka-console-producer.sh –broker-list localhost:9092 –topic test

(输入相应的消息,eg:hello kafka;按Ctrl+C结束)

6.启动Consumer并接收消息:

#./kafka-console-consumer.sh –zookeeper 127.0.0.1:2181 –topic test –from-beginning

7.前台启动kafka:

./kafka-server-start.sh ../config/server.properties

8.后台启动kafka:

./kafka-server-start.sh ../config/server.properties 1>/dev/null 2>&1 &

9.指定监听端口

JMX_PORT=2898

./kafka-server-start.sh ../config/server.properties 1>/dev/null 2>&1 &

 

二、Zookeeper常用操作

1.Zookeeper服务端启动:

# cd /opt/zookeeper-3.4.10/bin/

#./zkServer.sh start

2.Zookeeper服务端停止:

# cd /opt/zookeeper-3.4.10/bin/

#./zkServer.sh stop

3.Zookeeper服务端重启:

# cd /opt/zookeeper-3.4.10/bin/

#./zkServer.sh restart

4.查看Zookeeper进程:

#ps -ef|grep zookeeper;

5.查看Zookeeper服务端状态:

# cd /opt/zookeeper-3.4.10/bin/

#./zkServer.sh status

6.Zookeeper客户端登陆:

# cd /opt/zookeeper-3.4.10/bin/

#./zkCli.sh -server 127.0.0.1:2181

 

Kafka、SpringMVC整合例子

一、安装zookeeper

1.下载安装包:http://zookeeper.apache.org/releases.html#download;

2.进入Zookeeper设置目录,笔者D:\kafka\zookeeper-3.4.11\conf;

3. 将“zoo_sample.cfg”重命名为“zoo.cfg” ;

3. 编辑zoo.cfg配置文件;

4. 找到并编辑

dataDir=/tmp/zookeeper 并更改成您当前的路径;

5. 系统环境变量:

a. 在系统变量中添加ZOOKEEPER_HOME = D:\kafka\zookeeper-3.4.11

b. 编辑path系统变量,添加为路径%ZOOKEEPER_HOME%\bin;

6. 在zoo.cfg文件中修改默认的Zookeeper端口(默认端口2181);

7.打开新的cmd,输入zkServer,运行Zookeeper;

出现如下图片表示成功:

002

 

二、安装并运行Kafka

1.下载Kafka:http://kafka.apache.org/downloads.html

2. 进入Kafka配置目录,D:\kafka\kafka_2.12-1.0.1\config;

3. 编辑文件“server.properties” ;

4. 找到并编辑log.dirs=/tmp/kafka-logs 改成您当前可用的目录;

5. 找到并编辑zookeeper.connect=localhost:2181;

6. Kafka会按照默认,在9092端口上运行,并连接zookeeper的默认端口:2181。

运行Kafka代码:.\bin\windows\kafka-server-start.bat .\config\server.properties 

003

 

 

注:请确保在启动Kafka服务器前,Zookeeper实例已经准备好并开始运行。

三、Kafka代码的实现

1.生产者配置文件:

@Bean
public Map<String,Object> getDefaultFactoryArg(){
    Map<String,Object> arg = new HashMap<>();
    arg.put("bootstrap.servers",ConstantKafka.KAFKA_SERVER);
    arg.put("group.id","100");
    arg.put("retries","1");
    arg.put("batch.size","16384");
    arg.put("linger.ms","1");
    arg.put("buffer.memory","33554432");
    arg.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
    arg.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
    arg.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    arg.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
    return arg;
}

@Bean
public DefaultKafkaProducerFactory defaultKafkaProducerFactory(){
    DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory(this.getDefaultFactoryArg());
    return factory;
}

@Bean
public KafkaTemplate kafkaTemplate(){
    KafkaTemplate template = new KafkaTemplate(defaultKafkaProducerFactory());
    template.setDefaultTopic(ConstantKafka.KAFKA_TOPIC1);
    template.setProducerListener(kafkaProducerListener());
    return template;
}

@Bean
public KafkaProducerListener kafkaProducerListener(){
    KafkaProducerListener listener = new KafkaProducerListener();
    return listener;
}

2.消费者配置文件:

@Bean
public Map<String,Object> getDefaultArgOfConsumer(){
    Map<String,Object> arg = new HashMap<>();
    arg.put("bootstrap.servers",ConstantKafka.KAFKA_SERVER);
    arg.put("group.id","100");
    arg.put("enable.auto.commit","false");
    arg.put("auto.commit.interval.ms","1000");
    arg.put("auto.commit.interval.ms","15000");
    arg.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
    arg.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
    arg.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    arg.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

    return arg;
}

@Bean
public DefaultKafkaConsumerFactory defaultKafkaConsumerFactory(){
    DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(getDefaultArgOfConsumer());
    return factory;
}

@Bean
public KafkaConsumerMessageListener kafkaConsumerMessageListener(){
    KafkaConsumerMessageListener listener = new KafkaConsumerMessageListener();
    return listener;
}

/**
 * 监听频道-log
 * @return
 */
@Bean
public ContainerProperties containerPropertiesOfLog(){
    ContainerProperties properties = new ContainerProperties(ConstantKafka.KAFKA_TOPIC1);
    properties.setMessageListener(kafkaConsumerMessageListener());
    return properties;
}

/**
 * 监听频道-other
 * @return
 */
@Bean
public ContainerProperties containerPropertiesOfOther(){
    ContainerProperties properties = new ContainerProperties(ConstantKafka.KAFKA_TOPIC2);
    properties.setMessageListener(kafkaConsumerMessageListener());
    return properties;
}

@Bean(initMethod = "doStart")
public KafkaMessageListenerContainer kafkaMessageListenerContainerOfLog(){
    KafkaMessageListenerContainer container = new KafkaMessageListenerContainer(defaultKafkaConsumerFactory(),containerPropertiesOfLog());
    return container;
}

@Bean(initMethod = "doStart")
public KafkaMessageListenerContainer kafkaMessageListenerContainerOfOther(){
    KafkaMessageListenerContainer container = new KafkaMessageListenerContainer(defaultKafkaConsumerFactory(),containerPropertiesOfOther());
    return container;
}

3.生产消息服务

@Service
public class KafkaProducerServer implements IKafkaProducerServer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public static final String ROLE_LOG = "log";
    public static final String ROLE_web = "web";
    public static final String ROLE_APP = "app";

    /**
     * 发送消息
     * @param topic 频道
     * @param msg 消息对象
     * @param isUsePartition 是否使用分区
     * @param partitionNum 分区数,如果isUsePartition为true,此数值必须>0
     * @param role 角色:app,web
     * @return
     * @throws IllegalServiceException
     */
    @Override
    public ResultResp<Void> send(String topic, Object msg, boolean isUsePartition, Integer partitionNum, String role) throws IllegalServiceException {
        if (role == null){
            role = ROLE_LOG;
        }

        String key = role + "_" + msg.hashCode();
        String valueString = JsonUtil.ObjectToJson(msg, true);

        if (isUsePartition) {
            //表示使用分区
            int partitionIndex = getPartitionIndex(key, partitionNum);
            ListenableFuture<SendResult<String, Object>> result = kafkaTemplate.send(topic, partitionIndex, key, valueString);
            return checkProRecord(result);
        } else {
            ListenableFuture<SendResult<String, Object>> result = kafkaTemplate.send(topic, key, valueString);
            return checkProRecord(result);
        }
    }

    /**
     * 根据key值获取分区索引
     *
     * @param key
     * @param num
     * @return
     */
    private int getPartitionIndex(String key, int num) {
        if (key == null) {
            Random random = new Random();
            return random.nextInt(num);
        } else {
            int result = Math.abs(key.hashCode()) % num;
            return result;
        }
    }

    /**
     * 检查发送返回结果record
     *
     * @param res
     * @return
     */

    private ResultResp<Void> checkProRecord(ListenableFuture<SendResult<String, Object>> res) {
        ResultResp<Void> resp = new ResultResp<>();
        resp.setCode(ConstantKafka.KAFKA_NO_RESULT_CODE);
        resp.setInfo(ConstantKafka.KAFKA_NO_RESULT_MES);

        if (res != null) {
            try {
                SendResult r = res.get();//检查result结果集
                /*检查recordMetadata的offset数据,不检查producerRecord*/
                Long offsetIndex = r.getRecordMetadata().offset();
                if (offsetIndex != null && offsetIndex >= 0) {
                    resp.setCode(ConstantKafka.SUCCESS_CODE);
                    resp.setInfo(ConstantKafka.SUCCESS_MSG);
                } else {
                    resp.setCode(ConstantKafka.KAFKA_NO_OFFSET_CODE);
                    resp.setInfo(ConstantKafka.KAFKA_NO_OFFSET_MES);
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
                resp.setCode(ConstantKafka.KAFKA_SEND_ERROR_CODE);
                resp.setInfo(ConstantKafka.KAFKA_SEND_ERROR_MES + ":" + e.getMessage());

            } catch (ExecutionException e) {
                e.printStackTrace();
                resp.setCode(ConstantKafka.KAFKA_SEND_ERROR_CODE);
                resp.setInfo(ConstantKafka.KAFKA_SEND_ERROR_MES + ":" + e.getMessage());
            }
        }

        return resp;
    }

}

4.生产者监听服务

public class KafkaProducerListener implements ProducerListener {

    protected final Logger logger = Logger.getLogger(KafkaProducerListener.class.getName());

    public KafkaProducerListener(){

    }

    @Override
    public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
        logger.info("-----------------kafka发送数据成功");
        logger.info("----------topic:"+topic);
        logger.info("----------partition:"+partition);
        logger.info("----------key:"+key);
        logger.info("----------value:"+value);
        logger.info("----------RecordMetadata:"+recordMetadata);
        logger.info("-----------------kafka发送数据结束");
    }

    @Override
    public void onError(String topic, Integer partition, Object key, Object value, Exception e) {
        logger.info("-----------------kafka发送数据失败");
        logger.info("----------topic:"+topic);
        logger.info("----------partition:"+partition);
        logger.info("----------key:"+key);
        logger.info("----------value:"+value);
        logger.info("-----------------kafka发送数据失败结束");
        e.printStackTrace();
    }

    /**
     * 是否启动Producer监听器
     * @return
     */
    @Override
    public boolean isInterestedInSuccess() {
        return false;
    }
}

5.消费者监听服务

public class KafkaConsumerMessageListener implements MessageListener<String,Object> {

    private Logger logger = Logger.getLogger(KafkaConsumerMessageListener.class.getName());

    public KafkaConsumerMessageListener(){

    }

    /**
     * 消息接收-LOG日志处理
     * @param record
     */
    @Override
    public void onMessage(ConsumerRecord<String, Object> record) {
        logger.info("=============kafka消息订阅=============");

        String topic = record.topic();
        String key = record.key();
        Object value = record.value();
        long offset = record.offset();
        int partition = record.partition();

        if (ConstantKafka.KAFKA_TOPIC1.equals(topic)){
            doSaveLogs(value.toString());
        }

        logger.info("-------------topic:"+topic);
        logger.info("-------------value:"+value);
        logger.info("-------------key:"+key);
        logger.info("-------------offset:"+offset);
        logger.info("-------------partition:"+partition);
        logger.info("=============kafka消息订阅=============");
    }

    private void doSaveLogs(String data){
        SocketIOPojo<BikeLogPojo> logs = JsonUtil.JsonToObject(data.toString(),SocketIOPojo.class);
        /**
         * 写入到数据库中
         */
    }
}

 

测试图片:

004