Kafka集群MQ

1
      kafka默认队列中的信息保存7天，可以配置这个时间
2
      kafka队列信息能够一直向硬盘中保存(理论上没有大小限制)

1
log.dirs=D:/kafka/data

1
dataDir=D:/kafka/data

1
       <!--使用kafka依赖-->
2
        <dependency>
3
            <groupId>org.springframework.kafka</groupId>
4
            <artifactId>spring-kafka</artifactId>
5
        </dependency>
6
        <!--能够将java对象和json字符串相互转换的依赖-->
7
        <dependency>
8
            <groupId>com.google.code.gson</groupId>
9
            <artifactId>gson</artifactId>
10
        </dependency>

1
spring:
2
  kafka:
3
    bootstrap-servers: localhost:9092 #配置kafka服务器的位置
4
    # consumer.group-id是比不配置的信息，是组名
5
    consumer:
6
      group-id: csmall

1
//启动支持kafka
2
@EnableKafka
3
//为了测试kafka发送和接受消息的效果，利用SpringBoot自带的任务调度工具，周期的发送消息到kafka
4
@EnableScheduling
5
@SpringBootApplication
6
@EnableDubbo
7
public class CsmallCartWebapiApplication {
8

9
    public static void main(String[] args) {
10
        SpringApplication.run(CsmallCartWebapiApplication.class, args);
11
    }
12

13
}

1
package cn.tedu.csmall.cart.webapi.kafka;
2

3
import cn.tedu.csmall.commons.pojo.cart.model.CartTbl;
4
import com.google.gson.Gson;
5
import org.apache.commons.lang.math.RandomUtils;
6
import org.springframework.beans.factory.annotation.Autowired;
7
import org.springframework.kafka.core.KafkaTemplate;
8
import org.springframework.scheduling.annotation.Scheduled;
9
import org.springframework.stereotype.Component;
10

11
//周期性的向kafka发送消息
12
//Producer实例需要交给Spring容器管理
13
@Component
14
public class Producer {
15
    //KafkaTemplate<[主题的类型],[发送消息的类型]>
16
    @Autowired
17
    private KafkaTemplate<String,String> kafkaTemplate;
18

19
    int i=1;
20
    //设置每个10s中发送一次消息
21
    @Scheduled(fixedRate = 10000)
22
    public void sendMessage(){
23
        //实例化一个CartTbl对象，当做消息发送
24
        CartTbl cart = new CartTbl();
25
        cart.setId(i++);
26
        cart.setCommodityCode("PC100");
27
        cart.setPrice(RandomUtils.nextInt(300)+50);
28
        cart.setCount(RandomUtils.nextInt(10)+1);
29
        cart.setUserId("UU100");
30
        //将cart对象转成json格式的字符串
31
        Gson gson = new Gson();
32
        String json = gson.toJson(cart);
33
        System.out.println("本次发送的消息为："+json);
34
        //执行发送
35
        kafkaTemplate.send("myCart",json);
36
    }
37
}

1
@Component
2
public class Consumer {
3
    //SpringKafka接受消息依靠的是一个监听机制
4
    //SpringKafka提供了一个监听线程，实时关注kafka的消息往来
5
    //如果有消息发送到指定的主题，就会自动调用下面的方法
6
    @KafkaListener(topics = "myCart")
7
    public void received(ConsumerRecord<String,String> record){
8
       //ConsumerRecord<[主题类型],[消息类型]>
9
        //record是消息本身
10
        String json = record.value();
11
        //将json格式转成指定的java对象
12
        Gson gson = new Gson();
13
        CartTbl cart = gson.fromJson(json, CartTbl.class);
14
        System.out.println("接受到消息："+cart);
15

16
    }
17
}