SSM框架SpringMVC+ Spring + MyBatis（集成kafka消息）

超、凢脫俗 2023-02-09 13:11 9阅读 0赞

## 1.安装kafka的环境 ##

kafka的安装环境参考我的另外一篇博文 [https://blog.csdn.net/qq\_33371766/article/details/87866533][https_blog.csdn.net_qq_33371766_article_details_87866533]

安装完成并启动zookeeper和kafka。

## 2.ssm集成kafka ##

pom.xml:

<dependency>
    			<groupId>com.101tec</groupId>
    			<artifactId>zkclient</artifactId>
    			<version>0.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.zookeeper</groupId>
    			<artifactId>zookeeper</artifactId>
    			<version>3.4.5</version>
    		</dependency>
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>dubbo</artifactId>
    			<version>2.5.3</version>
    			<scope>compile</scope>
    			<exclusions>
    				<exclusion>
    					<artifactId>spring</artifactId>
    					<groupId>org.springframework</groupId>
    				</exclusion>
    			</exclusions>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.apache.kafka</groupId>
    			<artifactId>kafka-clients</artifactId>
    			<version>0.10.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    			<version>1.3.5.RELEASE</version>
    			<exclusions>
    				<exclusion>
    					<groupId>org.apache.kafka</groupId>
    					<artifactId>kafka-clients</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>

spring-kafka.xml:

<?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:context="http://www.springframework.org/schema/context"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
    	    http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
        <context:component-scan base-package="com.cxb.ssm.kafka"></context:component-scan>
    
        <bean id="kafkaProducerDemo" class="com.cxb.ssm.kafka.KafkaProducerDemo">
            <property name="properties">
                <props>
                    <prop key="topic">my-replicated-topic</prop>
                    <prop key="bootstrap.servers">127.0.0.1:9092</prop>
                    <prop key="acks">all</prop>
                    <prop key="key.serializer">org.apache.kafka.common.serialization.StringSerializer
                    </prop>
                    <prop key="value.serializer">org.apache.kafka.common.serialization.StringSerializer
                    </prop>
                    <prop key="buffer.memory">33554432</prop>
                </props>
    
            </property>
        </bean>
    
        <bean id="kafkaConsumerDemo" class="com.cxb.ssm.kafka.KafkaConsumerDemo">
            <property name="props">
                <props>
                    <prop key="topic">my-replicated-topic</prop>
                    <prop key="bootstrap.servers">127.0.0.1:9092</prop>
                    <prop key="group.id">group1</prop>
                    <prop key="enable.auto.commit">true</prop>
                    <prop key="auto.commit.interval.ms">1000</prop>
                    <prop key="session.timeout.ms">30000</prop>
                    <prop key="key.deserializer">org.apache.kafka.common.serialization.StringDeserializer
                    </prop>
                    <prop key="value.deserializer">org.apache.kafka.common.serialization.StringDeserializer
                    </prop>
                </props>
    
            </property>
        </bean>
    
    </beans>

KafkaConsumerDemo:

package com.cxb.ssm.kafka;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    public class KafkaConsumerDemo {
    
        private Properties props;
    
        public KafkaConsumerDemo(Properties props) {
            super();
            this.props = props;
        }
    
        public KafkaConsumerDemo() {
    
        }
    
        public Properties getProps() {
            return props;
        }
    
        public void setProps(Properties props) {
            this.props = props;
        }
    
        public String receive(){
    
            KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(props);
            consumer.subscribe(Arrays.asList(props.getProperty("topic")));
    
            String msg = "";
            while(true){
                ConsumerRecords<String,String> consumerRecords = consumer.poll(100);
                for(ConsumerRecord<String, String> consumerRecord:consumerRecords){
                    msg += consumerRecord.value();
                }
                consumer.close();
                return msg;
            }
        }
    
    }

KafkaProducerDemo:

package com.cxb.ssm.kafka;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class KafkaProducerDemo {
    
        Properties properties;
    
        public KafkaProducerDemo(Properties properties) {
            super();
            this.properties = properties;
        }
    
        public KafkaProducerDemo() {
            
        }
    
        public Properties getProperties() {
            return properties;
        }
    
        public void setProperties(Properties properties) {
            this.properties = properties;
        }
    
        public void sendMessage(String msg) {
    
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
    
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(properties.getProperty("topic"),
                    msg);
            producer.send(record);
    
            producer.close();
    
        }
    
    }

KafkaController:

package com.cxb.ssm.web;
    
    import com.cxb.ssm.kafka.KafkaConsumerDemo;
    import com.cxb.ssm.kafka.KafkaProducerDemo;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.*;
    import org.springframework.web.servlet.ModelAndView;
    
    import javax.annotation.Resource;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    @Controller
    public class KafkaController {
    
    	private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    
    	@Resource(name = "kafkaProducerDemo")
    	KafkaProducerDemo producer;
    
    	@Resource(name = "kafkaConsumerDemo")
    	KafkaConsumerDemo consumer;
    
    	@RequestMapping(value = "/welcome")
    	public ModelAndView welcome() {
    		logger.info("--------welcome--------");
    		ModelAndView mv = new ModelAndView();
    		mv.setViewName("welcome");
    		return mv;
    	}
    
    	@RequestMapping(value = "/sendmessage", method = RequestMethod.GET)
    	public ModelAndView sendMessage() {
    		logger.info("--------sendmessage--------");
    		Date date = new Date();
    		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    		String now = sdf.format(date);
    		ModelAndView mv = new ModelAndView();
    		mv.addObject("time", now);
    		mv.setViewName("send");
    		return mv;
    	}
    
    	@RequestMapping(value = "/onsend", method = RequestMethod.POST)
    	public ModelAndView onsend(@RequestParam("message") String msg) {
    		logger.info("--------onsend--------");
    		producer.sendMessage(msg);
    		ModelAndView mv = new ModelAndView();
    		mv.setViewName("welcome");
    		return mv;
    	}
    
    	@RequestMapping(value = "/receive")
    	public ModelAndView receive() {
    		logger.info("--------receive--------");
    		String msg = consumer.receive();
    		ModelAndView mv = new ModelAndView();
    		mv.addObject("msg", msg);
    		mv.setViewName("receive");
    		return mv;
    	}
    
    }

welcome.jsp

<%@ page language="java" contentType="text/html; charset=UTF-8"
             pageEncoding="UTF-8"%>
    <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
    <html>
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
        <title>welcome</title>
    </head>
    <body>
    <h1>Welcome</h1>
    <h2><a href="sendmessage">Send a Message</a></h2>
    <h2><a href="receive">Get a Message</a></h2>
    </body>
    </html>

send.jsp

<%@ page language="java" contentType="text/html; charset=UTF-8"
             pageEncoding="UTF-8"%>
    <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
    <html>
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
        <title>kafka_send</title>
    </head>
    <body>
    <h1>Send a Message</h1>
    <form action="onsend" method="post">
        MessageText:<textarea name="message">${time}</textarea>
        <br>
        <input type="submit" value="Submit">
    </form>
    
    <h2><a href="welcome">RETURN HOME</a></h2>
    
    </body>
    </html>

receive.jsp

<%@ page language="java" contentType="text/html; charset=UTF-8"
             pageEncoding="UTF-8"%>
    <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
    <html>
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
        <title>kafka_receive</title>
    </head>
    <body>
    
    <h1>Kafka_Reveive!!!</h1>
    <h2>Receive Message : ${msg}</h2>
    <h2><a href="welcome">RETURN HOME</a></h2>
    </body>
    </html>

测试：

发送消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70][]

收到的消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 1][]

测试成功。

代码下载： [https://github.com/ChenXbFrank/ssm/tree/devlop/][https_github.com_ChenXbFrank_ssm_tree_devlop]

[https_blog.csdn.net_qq_33371766_article_details_87866533]: https://blog.csdn.net/qq_33371766/article/details/87866533
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/202005182136315.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200518213645995.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2,size_16,color_FFFFFF,t_70
[https_github.com_ChenXbFrank_ssm_tree_devlop]: https://github.com/ChenXbFrank/ssm/tree/devlop/