0 条评论

使用spring+springMVC+mybatis+kafka做了两个web项目，一个是生产者，一个是消费者。

通过JMeter测试工具模拟100个用户并发访问生产者项目，发送json数据给生产者的接口，生产者将json数据发送到kafka集群，

消费者监听到kafka集群中的消息就开始消费，并将json解析成对象存到MySQL数据库。

下面是使用JMeter测试工具模拟100个并发的线程设置截图：

请求所发送的数据：

 

下面是100个用户10000个请求的聚合报告：

 

下面是生产者截图生产完10000条消息的时间截图：

 

下面是消费者项目消费入库的结束时间截图：

 

可见，10000条消息从生产完成到入库（消费完10000条消息的时间只是比生产完成的时间落后了几十秒，但是消费端真正入库完成所需要的时间很长）完成时间相差了10几分钟。

 

下面是MySQL数据库截图，数据全部入库成功：

下面是消息对应的POJO：

package com.xuebusi.pojo;

public class TbPerson {
    private Long id;

    private String name;

    private Integer age;

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name == null ? null : name.trim();
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "TbPerson [id=" + id + ", name=" + name + ", age=" + age + "]";
    }
}

 

下面是生产端的逻辑：

package com.xuebusi.controller;

import com.alibaba.fastjson.JSON;
import com.xuebusi.pojo.TbPerson;
import com.xuebusi.service.KafkaService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;

@Controller
@RequestMapping("/producer")
public class KafkaController {

    private static final Logger logger = LoggerFactory.getLogger(KafkaController.class);

    @Resource
    private KafkaService kafkaService;

    /**
     * 发消息到ssmk这个topic
     * @param person
     * @return
     */
    @RequestMapping(value = "/person", method = RequestMethod.POST)
    @ResponseBody
    public String createPerson(@RequestBody TbPerson person) {
        if (person == null){
            return "fail, data can not be null.";
        }
        String json = JSON.toJSONString(person);
        boolean result = kafkaService.sendInfo("ssmk", json);
        logger.info("生产者发送消息[" + result + "]:" + json);
        return "success";
    }
}

 

下面是消费端的逻辑：

package com.xuebusi.consumer;

import com.alibaba.fastjson.JSON;
import com.xuebusi.pojo.TbPerson;
import com.xuebusi.service.PersonService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

@Service
public class KafkaConsumerService {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    @Autowired
    private PersonService personService;

    public void processMessage(Map<String, Map<Integer, String>> msgs) {
        /*for (Map.Entry<String, Map<Integer, String>> entry : msgs.entrySet()) {
            String topic = entry.getKey();
            Map<Integer, String> value = entry.getValue();
            for (Map.Entry<Integer, String> entrySet : value.entrySet()) {
                Integer partiton = entrySet.getKey();
                String msg = entrySet.getValue();
                logger.info("消费的主题:" + topic + ",消费的分区:" + partiton + ",消费的消息:" + msg);
                logger.info("=======使用JSON解析对象=========");
                TbPerson person = JSON.parseObject(msg, TbPerson.class);
                logger.info("=======对象开始入库=========");
                personService.insert(person);
                logger.info("=======对象入库成功=========");
            }
        }*/

        for (Map.Entry<String, Map<Integer, String>> entry : msgs.entrySet()) {
            String topic = entry.getKey();
            Map<Integer, String> value = entry.getValue();
            for (Map.Entry<Integer, String> entrySet : value.entrySet()) {
                Integer partiton = entrySet.getKey();
                String msg = entrySet.getValue();
                logger.info("消费的主题:" + topic + ",消费的分区:" + partiton + ",消费的消息:" + msg);
                msg = "[" + msg + "]";//注意这里要在前后都加上中括号，否则下面在解析json成对象的时候会报json格式不对的异常（spring会对多条json数据用逗号分隔）
                logger.info("=======使用JSON解析对象=========");
                List<TbPerson> personList = JSON.parseArray(msg, TbPerson.class);
                //TbPerson person = JSON.parseObject(msg, TbPerson.class);
                if (personList != null && personList.size() > 0) {
                    logger.info("消息中包含[" + personList.size() + "]个对象");
                    for (TbPerson person : personList) {
                        logger.info("=======对象开始入库=========");
                        personService.insert(person);
                        logger.info("=======对象入库成功=========");
                    }
                }

            }
        }
    }
}

 

如果觉得本文对您有帮助，不妨扫描下方微信二维码打赏点，您的鼓励是我前进最大的动力：