kafka+java

写在前面

在之前的两篇文章中，我们介绍了log->logstash->kafka的流程连通，以及相关的环境搭建，现在准备工作都做好了，我们开始从kafka接收错误日志来发送邮件吧。

工程准备

这里我们搭建两个应用：

• kafka-spring ：用来监听kafka服务，判断错误系统，发送邮件给对应系统的负责人。
• MessageDispatcher ：用来发送消息，目前只只支持MAIL，后面会陆续加入SMS等。

kaka-spring

1.工程目录结构

kaka-spring.png

2.重要文件

重点看下applicationContext-consumer.xml和KafkaConsumerService这两个文件

• applicationContext-consumer.xml
  这个文件用来连接kafka和spring,看下具体内容

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
       xmlns:task="http://www.springframework.org/schema/task"
       xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
      http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
      http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
      http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
    <int:channel id="inputFromKafka">
        <int:queue/>
    </int:channel>

    <!--<int:service-activator auto-startup="true"-->
    <!--input-channel="inputFromKafka" ref="kafkaConsumerService" method="receiveMessage">-->
    <!--</int:service-activator>-->
    <!-- 上下两种方式都可以 -->
    <!-- 使用kafkaConsumerService来接收kafka消息 -->
    <int:outbound-channel-adapter channel="inputFromKafka"
                                  ref="kafkaConsumerService" method="receiveMessage" auto-startup="true"/>

    <int:poller default="true" id="default" fixed-rate="5"
                time-unit="MILLISECONDS" max-messages-per-poll="5">
    </int:poller>
    <int-kafka:inbound-channel-adapter
            kafka-consumer-context-ref="consumerContext" channel="inputFromKafka">
    </int-kafka:inbound-channel-adapter>
    <bean id="consumerProperties" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
        <property name="properties">
            <props>
                <prop key="auto.offset.reset">smallest</prop>
                <prop key="socket.receive.buffer.bytes">10485760</prop>
                <!-- 10M -->
                <prop key="fetch.message.max.bytes">5242880</prop>
                <prop key="auto.commit.interval.ms">1000</prop>
            </props>
        </property>
    </bean>
    <int-kafka:consumer-context id="consumerContext"
                                consumer-timeout="4000" zookeeper-connect="zookeeperConnect"
                                consumer-properties="consumerProperties">
        <int-kafka:consumer-configurations>
            <int-kafka:consumer-configuration
                    group-id="mygroup" max-messages="5000">
                <!-- 这里的topic就是我们再kafka中创建的那个 -->
                <int-kafka:topic id="kafkatopic" streams="4"/>
            </int-kafka:consumer-configuration>
        </int-kafka:consumer-configurations>
    </int-kafka:consumer-context>
    <!-- zookeeper地址按照自己的地址配置 -->
    <int-kafka:zookeeper-connect id="zookeeperConnect"
                                 zk-connect="192.168.1.120:2181" zk-connection-timeout="6000"
                                 zk-session-timeout="400" zk-sync-time="200"/>
</beans>

• KafkaConsumerService.java
  好，我们再来看下kafkaConsumerService中的receiveMessage方法是如何实现的。

public void receiveMessage(HashMap map)
    {
        logger.info("received Messages from kafka ================" + map.size());
        Set<Map.Entry> set = map.entrySet();

        for (Map.Entry entry : set)
        {
            String topic = (String) entry.getKey();
            logger.info("Topic:" + topic);
            ConcurrentHashMap<Integer, List<byte[]>> messages = (ConcurrentHashMap<Integer, List<byte[]>>) entry
                    .getValue();
            Collection<List<byte[]>> values = messages.values();

            for (Iterator<List<byte[]>> iterator = values.iterator(); iterator.hasNext(); )
            {
                List<byte[]> list = iterator.next();
                for (byte[] object : list)
                {
                    String message = new String(object);
                    Message msg = JSON.parseObject(message.replace("@", ""), Message.class);
                    String address = addressStore.pick(msg);
                    Map<String, String> request = new HashMap<String, String>();
                    request.put("messageType", Constants.MESSAGE_TYPE_MAIL);
                    request.put("address", address);
                    request.put("content", msg.toString());

                    try
                    {
                        //httpclient调用发送邮件
                        HttpClientUtil.postParameters(postUrl, request);
                    }
                    catch (Exception e)
                    {
                        logger.error("HttpClientUtil postParameters exception :" + e.getMessage());
                    }
                }
            }
        }
    }

3.重要的依赖

使用spring-integration项目中的kafka连接器

<dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-kafka</artifactId>
      <version>1.1.0.RELEASE</version>
</dependency>

其他的依赖就不一一写啦，相信大家都能搞定~以上就是kafka-spring的接收kafka消息的配置和java代码，其他的功能就在这个基础上加上去就可以了。

MessageDispatcher

1.目录结构

MessageDispatcher.png

典型的spring mvc项目，用作消息的分发。

2.重要文件

消息发送顶层接口MessageSender，不同的消息类型按照自己的要求实现接口的sendMessage方法即可。

package com.allinpay.message;

/**
 * 消息发送接口
 * Created by gejunqing on 16/9/8.
 */
public interface MessageSender
{
    void sendMessage(String address, String content);
}

再看看mvc的入口MessageController

package com.allinpay.mvc;

import com.allinpay.base.Constants;
import com.allinpay.base.MessageRequest;
import com.allinpay.base.SpringContextHolder;
import com.allinpay.message.MessageSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.ModelAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.servlet.ModelAndView;
import org.springframework.web.servlet.view.json.MappingJackson2JsonView;

import java.util.HashMap;
import java.util.Map;

@Controller
@RequestMapping("/api")
public class MessageController
{
    private Logger logger = LoggerFactory.getLogger(MessageController.class);


    @RequestMapping(value = "/sendMessage", method = RequestMethod.POST)
    public ModelAndView sendMessage(@ModelAttribute("request") MessageRequest request, ModelMap model)
    {
        logger.info("sendMessage {}", request);
        Map<String, String> result = new HashMap<String, String>();
        try
        {
            MessageSender sender = SpringContextHolder.getBean(request.getMessageType());
            sender.sendMessage(request.getAddress(), request.getContent());
            result.put("retCode", Constants.RET_CODE_SUCCESS);
        }
        catch (Exception e)
        {
            logger.error(e.getMessage());
            result.put("retCode", Constants.RET_CODE_SYTEMERROR);
            result.put("retMsg", e.getMessage());
        }
        return new ModelAndView(new MappingJackson2JsonView(), result);
    }
}

其他系统通过http post方式请求/api/sendMessage方式即可调用发送消息功能。

总结

至此，我们整个监控日志，发送邮件的流程已经全部结束。