投稿
  1. 首页
  2. PHP

实用开发篇-SpringBoot整合第三方技术

code • 2022-4-28 • PHP • 阅读 96

实用开发篇-SpringBoot整合第三方技术,第1张

SpringBoot整合第三方技术 1.缓存缓存使用缓存使用案例——手机验证码缓存供应商变更：Ehcache缓存供应商变更：Redis缓存供应商变更：memcached缓存供应商变更：jetcache缓存供应商变更：j2cache 2.任务Spring Task 3.邮件SpringBoot整合JavaMail 发送简单邮件 4.消息消息案例——订单短息通知ActiveMQActiveMQ服务启动（控制台）SpringBoot整合ActiveMQ RabbitMQRabbitMQ环境依赖配置，启动SpringBoot 整合RabbitMQ直连交换机模式SpringBoot 整合RabbitMQ主题交换机模式 RocketMQRocketMQ的环境依赖配置，启动SpringBoot整合RocketMQ kafkakafka运行启动SpringBoot整合Kafka

1.缓存 缓存是一种介于数据永久存储介质与数据应用之间的数据临时存储介质使用缓存可以有效的减少低速数据读取过程的次数（例如磁盘IO），提高系统性能缓存不仅可以用于提高永久性存储介质的数据读取效率，还可以提供临时的数据存储空间SpringBoot提供了缓存技术，方便缓存使用启动缓存设置进入缓存的数据设置读取缓存的数据 缓存使用 导入缓存技术对应的starter 
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-cacheartifactId>
        dependency>
启用缓存 
@SpringBootApplication
//开启缓存功能
@EnableCaching
public class Springboot19CacheApplication {
	public static void main(String[] args) {
		SpringApplication.run(Springboot19CacheApplication.class, args);
	}

}
设置当前 *** 作的结果数据进入缓存 
    @Cacheable(value = "cacheSpace", key = "#id")
    public Book getById(Integer id) {
        return bookDao.selectById(id);
    }
SpringBoot提供的缓存技术除了提供默认的缓存方案，还可以对其他缓存技术进行整合，统一接口，方便缓存技术的开发与管理
          ♦ Generic
          ♦ JCache
          ♦ Ehcache
          ♦ Hazelcast
          ♦ Infinispan
          ♦ Couchbase
          ♦ Redis
          ♦ Caffenine
          ♦ Simple(默认)
          ♦ memcached
          ♦ jetcache(阿里) 缓存使用案例——手机验证码 需求
          ♦ 输入手机号获取验证码，组织文档以短信形式发送给用户（页面模拟）
          ♦ 输入手机号和验证码验证结果需求分析
          ♦ 提供controller,传入手机号，业务层通过手机号计算出独有的6位验证码数据，存入缓存后返回此数据
          ♦ 提供controller，传入手机号与验证码，业务通过手机号从缓存中读取验证码与输入验证码进行比对，返回比对结果 缓存供应商变更：Ehcache 加入Ehcache坐标（缓存供应商实现） 
        <dependency>
            <groupId>net.sf.ehcachegroupId>
            <artifactId>ehcacheartifactId>
        dependency>
缓存设定为使用Ehcache 
spring:
  cache:
    type: ehcache
     ehcarche:
      config: ehcache.xml
提供ehcache配置文件ehcache.xml 

<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="http://ehcache.org/ehcache.xsd"
         updateCheck="false">
    <diskStore path="D:\ehcache" />

    
    
    
    
    
    
    
    
    <defaultCache
        eternal="false"
        diskPersistent="false"
        maxElementsInMemory="1000"
        overflowToDisk="false"
        timeToIdleSeconds="60"
        timeToLiveSeconds="60"
        memoryStoreEvictionPolicy="LRU" />

    <cache
        name="smsCode"
        eternal="false"
        diskPersistent="false"
        maxElementsInMemory="1000"
        overflowToDisk="false"
        timeToIdleSeconds="10"
        timeToLiveSeconds="10"
        memoryStoreEvictionPolicy="LRU" />

ehcache>
缓存供应商变更：Redis 加入Redis坐标（缓存供应商实现） 
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-data-redisartifactId>
        dependency>
配置Redis服务器，缓存设定为使用Redis 
Spring:
 redis:
  host: localhost
  port: 6379
 cache:
  type: redis
  redis:
   use-key-prefix: true		 			#是否使用前缀名（系统定义前缀名）
   cache-null-values: false				#是否允许空值
   key-prefix: aa						#追加自定义前缀名
   time-to-live: 10s					#有效时间
缓存供应商变更：memcached 安装：memcached
          ♦ 使用管理员身份运行cmd指令
          ♦ 安装 
memcached.exe -d install
运行 memcached
          ♦ 启动服务 
memcached.exe -d start

          ♦ 停止服务

memcached.exe -d stop

memcached客户端选择
          ♦ Memcached Client for Java:最早期客户端，稳定可靠，用户群广
          ♦ SpyMemcached:效率更高
          ♦ Xmemcached:并发处理更好

SpringBoot未提供对memcached的整合，需要使用硬编码方式实现客户端初始化管理

加入Xmemcache坐标（缓存供应商实现）

        <dependency>
            <groupId>com.googlecode.xmemcachedgroupId>
            <artifactId>xmemcachedartifactId>
            <version>2.4.7version>
        dependency>
配置memcache服务器必要属性 
memcached:
  # memcache服务器地址
  servers: localhost:11211
  # 连接池的数量
  poolSize: 10
  # 设置默认 *** 作超时
  opTimeout:  3000
创建读取属性配置信息类，加载配置 
@Component
@ConfigurationProperties(prefix = "memcached")
@Data
public class XMemcachedProperties {
    private String servers;
    private int poolSize;
    private long opTimeout;

}
创建客户端配置类 
@Configuration
public class XMemcachedConfig {
    @Autowired
    private XMemcachedProperties memcachedProperties;
    @Bean
    public MemcachedClient getMemcachedClient() throws IOException {
        MemcachedClientBuilder memcachedClientBuilder = new XMemcachedClientBuilder(memcachedProperties.getServers());
        memcachedClientBuilder.setConnectionPoolSize(memcachedProperties.getPoolSize());
        memcachedClientBuilder.setOpTimeout(memcachedClientBuilder.getOpTimeout());

        MemcachedClient memcachedClient = memcachedClientBuilder.build();
        return memcachedClient;
    }
}
配置memcache属性 
@RestController
@RequestMapping("/sms")
public class SMSCodeController {
    @Autowired
    private MemcachedClient memcachedClient;

    @GetMapping
    public String getCode(String tele) {
        String code = smsCodeService.sendCodeToSMS(tele);
        //将数据加入memcache
        try {
            memcachedClient.set(tele, 10, code);
        } catch (Exception e) {

        }
        return code;
    }

    @PostMapping
    public boolean checkCode(SMSCode smsCode) {
        String code = null;
        try {
            code = memcachedClient.get(smsCode.getTele()).toString();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return smsCode.getCode().equals(code);
    }
}
缓存供应商变更：jetcache jetCache对SpringCache进行了封装，在原有的功能基础上实现了多级缓存、缓存统计、自动刷新、异步调用、数据报表等功能jetCache设定了本地缓存与远程缓存的多级缓存解决方案
          ♦ 本地缓存（local）
                   □ LinkedHashMap
                   □ Caffeine
          ♦ 远程缓存（remote）
                   □ Redis
                   □ Tair导入jetcache坐标 
        <dependency>
            <groupId>com.alicp.jetcachegroupId>
            <artifactId>jetcache-starter-redisartifactId>
            <version>2.6.4version>
        dependency>
配置远程缓存必要属性 
jetcache:
  remote:
    default:
      type: redis
      host: localhost
      port: 6379
      poolConfig:
        maxTotal: 50
    sms:
      type: redis
      host: localhost
      port: 6379
      poolConfig:
        maxTotal: 50
配置本地缓存必要属性 
jetcache:
  local:
    default:
      type: linkedhashmap
      keyConvertor: fastjson
配置范例 
jetcache:
  statIntervalMinutes: 15
  areaInCacheName: false
  local:
    default:
      type: linkedhashmap
      keyConvertor: fastjson
      limit: 100
  remote:
    default:
      host: localhost
      port: 6379
      type: redis
      keyConvertor: fastjson
      valueEncoder: java
      valueDecoder: java
      poolConfig:
        minIdle: 5
        maxIdle: 20
        maxTotal: 50
配置属性说明
属性默认值说明
jetcache.statIntervalMinutes0统计间隔，0表示不统计
jetcache.hiddenPackages无自动生成name时，隐藏指定的包名前缀
jetcache.[local|remote].${area}.type无缓存类型，本地支持linkedhashmap、caffeine，远程支持redis、tair
jetcache.[local|remote].${area}.keyConvertor无key转换器，当前仅支持fastjson
jetcache.[local|remote].${area}.valueEncoderjava仅remote类型的缓存需要指定，可选java和kryo
jetcache.[local|remote].${area}.valueDecoderjava仅remote类型的缓存需要指定，可选java和kryo
jetcache.[local|remote].${area}.limit100仅local类型的缓存需要指定，缓存实例最大元素数
jetcache.[local|remote].${area}.expireAfterWritelnMillis无穷大默认过期时间，毫秒单位
jetcache.local.${area}.expireAfterWritelnMillis0仅local类型的缓存有效，毫秒单位，最大不活动间隔
开启jetcache注解支持 
@SpringBootApplication
//jetCache启用缓存的主开关
@EnableCreateCacheAnnotation
public class Springboot20CacheApplication {
    public static void main(String[] args) {
        SpringApplication.run(Springboot20CacheApplication.class, args);
    }

}
声明缓存对象 
@Service
public class SMSCodeServiceImpl implements SMSCodeService {
    @Autowired
    private CodeUtils codeUtils;
    @CreateCache(name = "jetCache_", expire = 1000, timeUnit = TimeUnit.SECONDS, cacheType = CacheType.LOCAL)
    private Cache<String, String> jetCache;
    
}
*** 作缓存 
@Service
public class SMSCodeServiceImpl implements SMSCodeService {
    @Override
    public String sendCodeToSMS(String tele) {
        String code = codeUtils.generator(tele);
        jetCache.put(tele, code);
        return code;
    }

    @Override
    public boolean checkCode(SMSCode smsCode) {
        String code = jetCache.get(smsCode.getTele());
        return smsCode.getCode().equals(code);
    }


}
启动方法注解 
@SpringBootApplication
//jetCache启用缓存的主开关
@EnableCreateCacheAnnotation
//开启方法注解缓存
@EnableMethodCache(basePackages = "com.itheima")
public class Springboot20CacheApplication {
    public static void main(String[] args) {
        SpringApplication.run(Springboot20CacheApplication.class, args);
    }
}
使用方法注解 *** 作缓存 
@Service
public class BookServiceImpl implements BookService {
    @Autowired
    private BookDao bookDao;
    @Override
    @Cached(name = "book_", key = "#id", expire = 3600, cacheType = CacheType.LOCAL)
    //@CacheRefresh(refresh = 10)
    public Book getById(Integer id) {
        return bookDao.selectById(id);
    }
    
    @Override
    @CacheUpdate(name = "book_", key = "#book.id", value = "#book")
    public boolean update(Book book) {
        return bookDao.updateById(book) > 0;
    }

    @Override
    @CacheInvalidate(name = "book_", key = "#id")
    public boolean delete(Integer id) {
        return bookDao.deleteById(id) > 0;
    }

}
缓存对象必须保障可序列化 
@Data
public class Book implements Serializable {
}

jetcache:
  remote:
    default:
      host: localhost
      port: 6379
      keyConvertor: fastjson
      valueEncode: java
      valueDecode: java
      type: redis
查看缓存统计报告 
jetcache:
  statIntervalMinutes: 1
缓存供应商变更：j2cache j2cache是一个缓存整合框架，可以提供缓存的整合方案，使各种缓存搭配使用，自身不提供缓存功能基于 ehcache + redis 进行整合加入j2cache坐标，加入整合缓存的坐标 
        <dependency>
            <groupId>net.oschina.j2cachegroupId>
            <artifactId>j2cache-coreartifactId>
            <version>2.8.5-releaseversion>
        dependency>

        <dependency>
            <groupId>net.oschina.j2cachegroupId>
            <artifactId>j2cache-spring-boot2-starterartifactId>
            <version>2.8.0-releaseversion>
        dependency>

        <dependency>
            <groupId>net.sf.ehcachegroupId>
            <artifactId>ehcacheartifactId>
        dependency>
配置使用j2cache(application.yml) 

j2cache:
  config-location: j2cache.properties
配置一级缓存与二级缓存以及一级缓存数据到二级缓存的发送方式（j2cache.properties） 
# 配置1级缓存
j2cache.L1.provider_class=ehcache
ehcache.configXml=ehcache.xml

# 设置是否启用2级缓存
j2cache.l2-cache-open = false

# 配置1级缓存数据到2级h缓存的广播方式：可以使用redis提供的消息订阅模式，也可以使用jgroups多播实现
j2cache.broadcast=net.oschina.j2cache.cache.support.redis.SpringRedisPubSubPolicy

# 配置2级缓存
j2cache.L2.provider_class=net.oschina.j2cache.cache.support.redis.SpringRedisProvider
j2cache.L2.config_section=redis
redis.hosts=localhost:6379


redis.mode=single 
redis.namespace=j2cache
设置使用缓存 
@Service
public class SMSCodeServiceImpl implements SMSCodeService {
    @Autowired
    private CodeUtils codeUtils;

    @Autowired
    private CacheChannel cacheChannel;

    @Override
    public String sendCodeToSMS(String tele) {
        String code = codeUtils.generator(tele);
        cacheChannel.set("sms", tele, code);
        return code;
    }

    @Override
    public boolean checkCode(SMSCode smsCode) {
        String code = cacheChannel.get("sms", smsCode.getTele()).asString();
        return smsCode.getCode().equals(code);
    }


}
2.任务 定时任务是企业级应用中的常见 *** 作
          ♦ 年度报表
          ♦ 缓存统计报告市面上流行的定时任务技术
          ♦ Quartz
          ♦ Spring Task相关概念
          ♦ 工作（Job）：用于定义具体执行的工作
          ♦ 工作明细（JobDetail）：用于描述定时工作相关的信息
          ♦ 触发器（Tirgger）：用于描述触发工作的规则，通常使用cron表达式定义调度规则
          ♦ 调度器（Scheduler）：描述了工作明细与触发器之间的对应关系导入SpringBoot整合quartz的坐标 
		<dependency>
			<groupId>org.springframework.bootgroupId>
			<artifactId>spring-boot-starter-quartzartifactId>
		dependency>
定义具体要执行的任务，继承QuartzJobBean 
public class MyQuartz extends QuartzJobBean {
    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {

        System.out.println("quartz task run...");
    }
}
定义工作明细与触发器，并绑定对应关系 
@Configuration
public class QuartzConfig {
    @Bean
    public JobDetail printJobDetail() {
        //绑定具体的工作
        return JobBuilder.newJob(MyQuartz.class).storeDurably().build();
    }

    @Bean
    public Trigger printJobTrigger() {
        ScheduleBuilder schedBuilder = CronScheduleBuilder.cronSchedule("0/5 * * * * ?");
        //绑定对应的工作明细
        return TriggerBuilder.newTrigger().forJob(printJobDetail()).withSchedule(schedBuilder).build();

    }
}
Spring Task 开启定时任务功能 
@SpringBootApplication
//开启定时任务功能
@EnableScheduling
public class Springboot22TaskApplication {

    public static void main(String[] args) {
        SpringApplication.run(Springboot22TaskApplication.class, args);
    }

}
设置定时执行的任务，并设定执行周期 
@Component
public class MyBean {
    @Scheduled(cron = "0/1 * * * * ?")
    public void print() {
        System.out.println("spring task run.");
    }
}
定时任务相关配置 
spring:
  task:
    scheduling:
      # 任务调度线程池大小 默认1
     pool:
       size: 1
     # 调度线程名称前缀 默认scheduling-
     thread-name-prefix: ssm_
     shutdown:
       # 线程池关闭时等待所有任务完成
       await-termination: false
       # 调度线程关闭前最大等待时间，确保最后一定关闭
       await-termination-period: 10s
3.邮件 SpringBoot整合JavaMail 发送简单邮件

SMTP（Simple Mail Transfer Protocol）：简单邮件传输协议，用于发送电子邮件的传输协议

POP3（Post Office Protocol - Version 3）： 用于接收电子邮件的标准协议

IMAP（Internet Mail Access Protocol）：互联网消息协议，是POP3的替代协议

导入SpringBoot整合JavaMail坐标

	<dependency>
			<groupId>org.springframework.bootgroupId>
			<artifactId>spring-boot-starter-mailartifactId>
		dependency>
	dependencies>
配置JavaMail 
spring:
  mail:
    host: smtp.qq.com
    username: ****353464@qq.com
    password: *********

开启定时任务功能 
@Service
public class SendMailServiceImpl implements SendMailService {

    @Autowired
    private JavaMailSender javaMailSender;

    //发送人
    private String from = "[email protected]";
    //接收人
    private String to = "[email protected]";
    //标题
    private String subject = "测试邮件";
    //正文
    private String context = "测试邮件正文内容";

    @Override
    public void sendMail() {
        SimpleMailMessage message = new SimpleMailMessage();
        message.setFrom(from + "(小甜甜)");
        message.setTo(to);
        message.setSubject(subject);
        message.setText(context);
        javaMailSender.send(message);
    }
}
4.消息 消息发送方
          ♦ 生产者消息接收方
          ♦ 消费者同步消息异步消息
企业级应用中广泛使用的三种异步消息传递技术
          ♦ JMS：
                   □ JMS（Java Message Service）：一个规范，等同于JBDC规范，提供了与消息服务相关的API接口
                   □ JMS消息模型
                                 ⚪ peer-2-peer:点对点模型，消息发送到一个队列中，队列保存消息。队列消息只能被一个消费者消费，或超时
                                 ⚪ publish-subscribe:发布订阅模型，消息可以被多个消费者消费，生产者和消费者完全独立，不需要感知对方的存在
                   □ JMS消息种类
                                 ⚪TextMessage
                                 ⚪MapMessage
                                 ⚪BytesMessage
                                 ⚪StreamMessage
                                 ⚪ObjectMessage
                                 ⚪Message(只有消息头和属性)
                   □ JMS实现:ActiveMQ、Redis、HornetMQ、RabbitMQ、RocketMQ（没有完全遵守JMS规范）
          ♦ AMOP：
                   □ AMQP（advanced message queuing protocol）:一种协议（高级消息队列协议，也是消息代理规范），规范了网络交换的数据格式，兼容JMS
                   □ 优点：具有跨平台性，服务器供应商，生产者，消费者可以使用不同的语言来实现
                   □ AMQP消息模型
                                 ⚪ direct exchange
                                 ⚪ fanout exchange
                                 ⚪ topic exchange
                                 ⚪ headers exchange
                                 ⚪ system exchange
                   □ AMQP消息种类：byte[ ]
                   □ AMQP实现：RabbitMQ、StormMQ、RocketMQ
          ♦ MQTT（Message Queueing Telemetry Transport）消息队列遥测传输，专为小设备而设计，是物联网（IOT）生态系统中主要成分之一
          ♦ Kafka：一种高吞吐量的分布式订阅消息系统，提供实时消息功能 消息案例——订单短息通知 购物订单业务
          ♦ 登录状态检测
          ♦ 生成主单
          ♦ 生成子单
          ♦ 库存检测与变更
          ♦ 积分变更
          ♦ 支付
          ♦ 短信通知
          ♦ 购物车维护
          ♦ 运单信息初始化
          ♦ 商品库存维护
          ♦ 会员维护 ActiveMQ ActiveMQ服务启动（控制台） 启动服务 
activemq.bat
访问服务器 
http://127.0.0.1:8161/

          ♦ 服务器端口：61616，管理后台端口：8161
          ♦ 用户名&密码：admin

SpringBoot整合ActiveMQ 导入SpringBoot整合ActiveMQ坐标 
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-activemqartifactId>
        dependency>
配置ActiveMQ（采用默认配置） 
spring:
  activemq:
    broker-url: tcp://localhost:61616
  jms:
    template:
      default-destination: itheima
    pub-sub-domain: true
生产与消费消息（使用默认消息存储队列） 
@Service
public class MessageServiceActivemqImpl implements MessageService {

    @Autowired
    private JmsMessagingTemplate messageTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理的队列,id:" + id);
        messageTemplate.convertAndSend("order.queue.id:", id);

    }

    @Override
    public String doMessage() {
        String id = messageTemplate.receiveAndConvert("order.queue.id:", String.class);
        System.out.println("已完成短信发送业务,id" + id);
        return id;
    }
}
使用消息监听器对消息队列监听 

@Component
public class MessageListener {

    @JmsListener(destination = "order.queue.id:")
    @SendTo("order.other.queue.id:")
    public String receive(String id) {
        System.out.println("已完成短信发送业务,id" + id);
        return "new:" + id;
    }
}
RabbitMQ RabbitMQ环境依赖配置，启动

RabbitMQ基于Erlang语言编写，需要安装Erlang

Erlang
          ♦ 安装：一键安装，安装完毕需要重启，需要依赖windows组件

环境变量配置
          ♦ ERLANG_HOME
          ♦ PATH

启动服务

rabbitmq-service.bat start
关闭服务 
rabbitmq-service.bat stop
查看服务状态 
rabbitmqctl status
服务管理可视化（插件形式）
          ♦ 查看已安装的插件列表 
rabbitmq-plugins.bat list

          ♦ 开启服务管理插件

rabbitmq-plugins.bat enable rabbitmq_management

          ♦ 访问服务器

http://localhost:15672

                   □ 服务端口：5672，管理后台端口：15672
                   □ 用户名&密码：guest

SpringBoot 整合RabbitMQ直连交换机模式 导入SpringBoot整合RabbitMQ坐标 
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-amqpartifactId>
        dependency>
配置 RabbitMQ（采用默认配置） 
spring:
  rabbitmq:
    host: localhost
    port: 5672
定义消息队列（direct） 
@Configuration
public class RabbitConfigDirect {

    @Bean
    public Queue directQueue() {
    	// durable:是否持久化，默认false
    	// exclusive：是否当前连接专用，默认false，连接关闭后队列即被删除
    	// autoDelete：是否自动删除，当生产者或消费者不再使用此队列，自动给删除
        return new Queue("direct_queue",true,false,false);
    }
    @Bean
    public Queue directQueue2() {
        return new Queue("direct_queue2");
    }

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange");
    }

    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");
    }

    @Bean
    public Binding bindingDirect2() {
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("direct2");
    }
}
生产与消费消息（direct） 
@Service
public class MessageServiceRabbitmqDirectImpl implements MessageService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理的队列(rabbitmq direct),id:" + id);
        amqpTemplate.convertAndSend("directExchange", "direct", id);
    }
}
使用消息监听器对消息队列监听（direct） 
@Component
public class MessageListener {
    @RabbitListener(queues = "direct_queue")
    public void receive(String id) {
        System.out.println("已完成短信发送业务(RabbitMQ direct),id" + id);
    }
}
SpringBoot 整合RabbitMQ主题交换机模式 定义消息队列（topic） 
@Configuration
public class RabbitConfigTopic {

    @Bean
    public Queue topicQueue() {
        return new Queue("topic_queue");
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue("topic_queue2");
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }

    @Bean
    public Binding bindingTopic() {
        return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.*.id");
    }

    @Bean
    public Binding bindingTopic2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.orders.*");
    }
}
绑定键匹配规则
          ♦ *（星号）：用来表示一个单词，且该单词是必须出现的
          ♦ #（井号）：用来表示任意数量
匹配键topic.*.*topic.#
topic.order.idtruetrue
order.topic.idfalsefalse
topic.sm.order.idfalsetrue
topic.sm.idfalsetrue
topic.id.ordertruetrue
topic.idfalsetrue
topic.orderfalsetrue
生产与消费消息（topic） 
@Service
public class MessageServiceRabbitmqDirectImpl implements MessageService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理的队列(rabbitmq topic),id:" + id);
        amqpTemplate.convertAndSend("topicExchange", "topic.orders.id", id);
    }
}
使用消息监听器对消息队列监听（topic） 
@Component
public class MessageListener {
    @RabbitListener(queues = "topic_queue")
    public void receive(String id) {
        System.out.println("已完成短信发送业务(RabbitMQ topic 1),id" + id);
    }

    @RabbitListener(queues = "topic_queue")
    public void receive2(String id) {
        System.out.println("已完成短信发送业务(RabbitMQ topic 2),id" + id);
    }
}
RocketMQ RocketMQ的环境依赖配置，启动

默认服务器端口：9876

环境变量配置
          ♦ ROCKETMQ_HOME
          ♦ PATH
          ♦ NAMESRV_ADDR（建议）：127.0.0.1:9876

命名服务器与broker

启动命名服务器 
先启动NAMESERVER：start mqnamesrv.cmd
启动broker 
然后启动BROKER：start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

（备注：autoCreateTopicEnable=true 表示可以动态创建topic）

服务器功能测试：生产者 
tools.cmd  org.apache.rocketmq.example.quickstart.Producer
服务器功能测试：消费者 
tools.cmd  org.apache.rocketmq.example.quickstart.Consumer
SpringBoot整合RocketMQ 导入SpringBoot整合RocketMQ坐标 
        <dependency>
            <groupId>org.apache.rocketmqgroupId>
            <artifactId>rocketmq-spring-boot-starterartifactId>
            <version>2.2.2version>
        dependency>
配置RocketMQ（采用默认配置） 
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: group_rocketmq
生产消息（异步） 
@Service
public class MessageServiceRocketmqImpl implements MessageService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理的队列(rocketmq direct),id:" + id);
        //rocketMQTemplate.convertAndSend("order_id", id);
        SendCallback callback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("消息发送失败");
            }
        };
        rocketMQTemplate.asyncSend("order_id",id,callback);

    }
}
使用消息监听器对消息队列监听 
@Component
@RocketMQMessageListener(topic = "order_id",consumerGroup = "group_rocketmq")
public class MessageListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String id) {
        System.out.println("已完成短信发送业务(RocketMQ),id" + id);
    }
}
kafka kafka运行启动 把文件放在某盘下，cmd进入跟目录\bin\windows启动zookeeper 
zookeeper-server-start.bat ../../config/zookeeper.properites

          ♦ 默认端口：2181

启动kafka 
kafka-server-start.bat  ../../config/server.properties

          ♦ 默认端口：9092

创建topic 
kafka-topics.bat --zookeeper 127.0.0.1:2181 --create --replication-factor 1 --partitions 1 --topic itheima
查看topic 
kafka-topics.bat --zookeeper 127.0.0.1:2181 --list
删除topic 
kafka-topics.bat --zookeeper 127.0.0.1:2181 --delete --topic itheima
生产者功能测试 
kafka-console-producer.bat --broker-list localhost:9092 --topic itheima2022
消费者功能测试 
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic itheima2022 --from-beginning
SpringBoot整合Kafka 导入SpringBoot整合Kafka坐标 
        <dependency>
            <groupId>org.springframework.kafkagroupId>
            <artifactId>spring-kafkaartifactId>
        dependency>
配置Kafka（采用默认配置） 
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order
生产消息 
@Service
public class MessageServiceKafkalmpl implements MessageService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理的队列(kafka),id:" + id);
        kafkaTemplate.send("itheima2022", id);
    }
}
使用消息监听器对消息队列监听 
@Component
public class MessageServiceKafkaImpl {
    @KafkaListener(topics = {"itheima2022"})
    void onMessage(ConsumerRecord<String, String> record) {
        System.out.println("已完成短信发送业务(kafka),id" + record.value());
    }
}

欢迎分享，转载请注明来源：内存溢出

原文地址: https://outofmemory.cn/langs/737354.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-04-28
下一篇 2022-04-28

发表评论

登录后才能评论

评论列表（0条）

保存