spring整合kafka項目生產和消費測試結果記錄（一）

使用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數據用逗號分隔） 45                 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("=======對象入庫成功=========");
                    }
                }

            }
        }
    }
}

 

如果覺得本文對您有幫助，不妨掃描下方微信二維碼打賞點，您的鼓勵是我前進最大的動力：