这篇文章将为大家详细讲解有关如何实现Storm-kafka,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。
10年的合水网站建设经验,针对设计、前端、开发、售后、文案、推广等六对一服务,响应快,48小时及时工作处理。成都营销网站建设的优势是能够根据用户设备显示端的尺寸不同,自动调整合水建站的显示方式,使网站能够适用不同显示终端,在浏览器中调整网站的宽度,无论在任何一种浏览器上浏览网站,都能展现优雅布局与设计,从而大程度地提升浏览体验。创新互联从事“合水网站设计”,“合水网站推广”以来,每个客户项目都认真落实执行。
实现细节如下:
public static Iterable> generateTuples( KafkaConfig kafkaConfig, Message msg) { Iterable
> tups; ByteBuffer payload = msg.payload(); ByteBuffer key = msg.key(); if (key != null && kafkaConfig.scheme instanceof KeyValueSchemeAsMultiScheme) { tups = ((KeyValueSchemeAsMultiScheme) kafkaConfig.scheme) .deserializeKeyAndValue(Utils.toByteArray(key), Utils.toByteArray(payload)); } else { tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload)); } return tups; }
1 :方法内部 Iterable> 类型的tups对象
2 : Message对象是kafka的Message。 包括了Key,以及palyload负载
3 : 在生成元组的时候,需要拿到kafkaConfig.scheme 当中的Scheme,一旦Scheme instanceof
KeyValueSchemeAsMultiScheme,将强制性转换为 KeyValueSchemeAsMultiScheme,并且调用deserializeKeyAndValue 方法将其反序列化 List
细节上,让我们查看一下 KeyValueSchemeAsMultiScheme 对象
package com.mixbox.storm.kafka; import backtype.storm.spout.SchemeAsMultiScheme; import java.util.Arrays; import java.util.List; @SuppressWarnings("serial") public class KeyValueSchemeAsMultiScheme extends SchemeAsMultiScheme{ public KeyValueSchemeAsMultiScheme(KeyValueScheme scheme) { super(scheme); } //把一个key value 进行反序列化 public Iterable> deserializeKeyAndValue(final byte[] key, final byte[] value) { List
首先, KeyValueSchemeAsMultiScheme 继承自SchemeAsMultiScheme,SchemeAsMultiScheme是Storm自身所带有的一个class实现,以上的2个class 都实现了Storm之中【MultiScheme】
关于“如何实现Storm-kafka”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,使各位可以学到更多知识,如果觉得文章不错,请把它分享出去让更多的人看到。
分享文章:如何实现Storm-kafka
路径分享:http://lswzjz.com/article/gdoepe.html