/*** TODO* * @author smartloli.**         Created by Mar 9, 2018*/
public class KafkaSqlFactory {private static final Logger LOG = LoggerFactory.getLogger(KafkaSqlFactory.class);private static Ignite ignite = null;private static void getInstance() {if (ignite == null) {ignite = Ignition.start();}}private static IgniteCache<Long, TopicX> processor(List<TopicX> collectors) {getInstance();CacheConfiguration<Long, TopicX> topicDataCacheCfg = new CacheConfiguration<Long, TopicX>();topicDataCacheCfg.setName(TopicCache.NAME);topicDataCacheCfg.setCacheMode(CacheMode.PARTITIONED);topicDataCacheCfg.setIndexedTypes(Long.class, TopicX.class);IgniteCache<Long, TopicX> topicDataCache = ignite.getOrCreateCache(topicDataCacheCfg);for (TopicX topic : collectors) {topicDataCache.put(topic.getOffsets(), topic);}return topicDataCache;}public static String sql(String sql, List<TopicX> collectors) {try {IgniteCache<Long, TopicX> topicDataCache = processor(collectors);SqlFieldsQuery qry = new SqlFieldsQuery(sql);QueryCursor<List<?>> cursor = topicDataCache.query(qry);for (List<?> row : cursor) {System.out.println(row.toString());}} catch (Exception ex) {LOG.error("Query kafka topic has error, msg is " + ex.getMessage());} finally {close();}return "";}private static void close() {try {if (ignite != null) {ignite.close();}} catch (Exception ex) {LOG.error("Close Ignite has error, msg is " + ex.getMessage());} finally {if (ignite != null) {ignite.close();}}}}

public static void ignite(){List<TopicX> collectors = new ArrayList<>();int count = 0;for (int i = 0; i < 10; i++) {TopicX td = new TopicX();if (count > 3) {count = 0;}td.setPartitionId(count);td.setOffsets(i);td.setMessage("hello_" + i);td.setTopicName("test");collectors.add(td);count++;}String sql = "select offsets,message from TopicX where offsets>6 and partitionId in (0,1) limit 1";long stime = System.currentTimeMillis();KafkaSqlFactory.sql(sql, collectors);System.out.println("Cost time [" + (System.currentTimeMillis() - stime) / 1000.0 + "]ms");}

　　这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

联系方式：
邮箱：smartloli.org@gmail.com
Twitter：https://twitter.com/smartloli
QQ群（Hadoop - 交流社区1）：424769183
温馨提示：请大家加群的时候写上加群理由（姓名＋公司/学校），方便管理员审核，谢谢！

热爱生活，享受编程，与君共勉！

作者：哥不是小萝莉 ［关于我］［犒赏］

出处：http://www.cnblogs.com/smartloli/

转载请注明出处，谢谢合作！