微服务 Spring Cloud Alibaba 项目搭建（七、

作者: 王疏蔬 | 来源:发表于2021-04-14 18:09 被阅读0次

Spring Cloud Alibaba 微服务原理与实战
5、Spring Cloud Alibaba微服务系列-分布式锁
spring cloud alibaba系列(一)Nacos 服
Nacos 服务注册与发现实操
3，创建统一的依赖管理
Spring Cloud Alibaba
4.Spring Cloud Alibaba 创建服务提供者
4.Spring Cloud Alibaba 创建服务提供者
谷粒商城-基础篇- 笔记十一-（分布式组件-spring clo
Spring Cloud Alibaba-概要

RocketMQ介绍

RocketMQ 是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式的特点。它是一个采用 Java 语言开发的分布式的消息系统，由阿里巴巴团队开发，在2016年底贡献给 Apache，成为了 Apache 的一个顶级项目。在阿里内部，RocketMQ 很好地服务了集团大大小小上千个应用，在每年的双十一当天，更有不可思议的万亿级消息通过 RocketMQ 流转。

RocketMQ 特点

是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式等特点

Producer、Consumer、队列都可以分布式

Producer 向一些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做广播消费，则一个 Consumer 实例消费这个 Topic 对应的所有队列，如果做集群消费，则多个 Consumer 实例平均消费这个 Topic 对应的队列集合

能够保证严格的消息顺序

支持拉（pull）和推（push）两种消息模式

高效的订阅者水平扩展能力

实时的消息订阅机制

亿级消息堆积能力

支持多种消息协议，如 JMS、OpenMessaging 等

较少的依赖

kafka 、RocketMQ 、RabbitMQ 对比

RocketMQ安装

RocketMQ下载： rocketmq-all-4.8.0-bin-release.zip
1.RocketMQ zip包传入linux服务器

[root@localhost ]# cd usr/local/
[root@localhost local]# rz

2.解压缩

[root@localhost local]# unzip rocketmq-all-4.8.0-bin-release.zip

3.调整启动参数（修改默认启动参数，默认启动的最大内存为4G，比较大，修改小一点，否则如果服务器内存不够会启动失败）

[root@localhost local]# cd rocketmq-all-4.8.0-bin-release/bin
[root@localhost bin]# vim runserver.sh

-Xms4g -Xmx4g -Xmn2g 改为 -Xms256m -Xmx256m -Xmn128m

4.调整broker

[root@localhost bin]# vim runbroker.sh

-Xms8g -Xmx8g -Xmn4g 改为 -Xms256m -Xmx256m -Xmn128m

5.启动namesrv

[root@localhost bin]# nohup sh mqnamesrv &

6.启动broker,注意ip为公网ip,端口为navmesrv的默认端口9876

[root@localhost bin]# nohup ./mqbroker -n localhost:9876 &

7.检查是否启动成功

[root@localhost bin]# jps -l

如果发现报错bash: jps: 未找到命令... 请更新以下命令

[root@localhost bin]# sudo yum install java-1.8.0-openjdk-devel.x86_64

输入命令 jps -l

关闭 RocketMQ 命令（此处无需关闭，只用于了解）

./mqshutdown broker
./mqshutdown namesrv

RocketMQ 控制台安装

1.克隆rocketmq项目

[root@localhost local]# cd /usr/local/
[root@localhost local]# git clone  https://github.com/apache/rocketmq-externals.git

进入\rocketmq-externals\rocketmq-console\src\main\resources\ 下修改 application.properties 配置文件

配置文件修改如下图

github提供了 Docker 和非Docker 两种安装方法供其选择，这里使用非Docker方式进行安装

在 \rocketmq-externals\rocketmq-console\ 文件夹下打开控制台，输入以下命令进行maven打包

mvn clean package -Dmaven.test.skip=true

进入 \rocketmq-externals\rocketmq-console\target\ 文件夹下打开控制台，输入以下命令进行 jar包启动

java -jar rocketmq-console-ng-2.0.0.jar

打开浏览器访问 localhost:9877，如果报错

开放 10909 01911 9876 端口

firewall-cmd --zone=public --add-port=10909/tcp --permanent
firewall-cmd --zone=public --add-port=10911/tcp --permanent
firewall-cmd --zone=public --add-port=9876/tcp --permanent
systemctl restart firewalld.service
firewall-cmd --reload

验证RocketMQ功能是够正常

1.验证生产消息正常，输入命令

[root@localhost rocketmq-all-4.8.0-bin-release]# export NAMESRV_ADDR=localhost:9876
[root@localhost rocketmq-all-4.8.0-bin-release]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

正常的情况下，会看到一堆的类似于如下的输出，这是生产消息后成功的result：

SendResult [sendStatus=SEND_OK, msgId=7F000001372329453F44466341350068, offsetMsgId=C0A8017600002A9F000000000000674E, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=3], queueOffset=33]

2.验证消费消息正常，执行如下命令：

[root@localhost rocketmq-all-4.8.0-bin-release]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

正常的情况下，会看到一堆的类似于如下的输出，这是消费的消息内容：

ConsumeMessageThread_1 Receive New Messages: [MessageExt [brokerName=localhost.localdomain, queueId=0, storeSize=201, queueOffset=0, sysFlag=0, bornTimestamp=1618387294736, bornHost=/192.168.1.118:34722, storeTimestamp=1618387294743, storeHost=/192.168.1.118:10911, msgId=C0A8017600002A9F0000000000000192, commitLogOffset=402, bodyCRC=1250039395, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=34, CONSUME_START_TIME=1618387666005, UNIQ_KEY=7F00000136FE29453F44466306100001, CLUSTER=DefaultCluster, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 50], transactionId='null'}]]

访问页面 http://localhost:9877 ，出现下图界面，安装成功

RocketMQ 集成 - 生产者

gateway下pom.xml文件添加依赖

<!-- rocketmq -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>

nacos 配置 RocketMQ

rocketmq:
  name-server: 192.168.190.129:9876
  producer:
    # 小坑：必须指定group
    group: test-group

common 下创建实体类 MyMessage.class

package com.bi.cloud.pojo;

import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
public class MyMessage implements Serializable {
    private Integer id;
    private String name;
    private String status;
    private Date createTime;
}

gateway下创建 TestProducerController.class

package com.bi.cloud.controller;

import com.bi.cloud.pojo.MyMessage;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;

/**
 * 生产者
 **/
@RestController
@RequestMapping("/api/testRocketMQ")
public class TestProducerController {

    /**
     * 用于发送消息到 RocketMQ 的api
     */
    @Resource
    public RocketMQTemplate rocketMQTemplate;

    @GetMapping("/sendMsg")
    public String testSendMsg() {
        String topic = "test-topic";
        MyMessage message = new MyMessage();
        message.setId(1);
        message.setName("王霄");
        message.setStatus("default");
        message.setCreateTime(new Date());
        // 发送消息
        rocketMQTemplate.convertAndSend(topic, message);

        return "send message success";
    }
}

Postman 调用接口

如果报错请关闭linux防火墙

systemctl stop firewalld

消息发送成功后，可以到RocketMQ的控制台中进行查看：

RocketMQ 集成 - 消费者

engine下pom.xml文件添加依赖

<!-- rocketmq -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>

nacos 配置 RocketMQ

rocketmq:
  name-server: 192.168.190.129:9876
  producer:
    # 小坑：必须指定group
    group: test-group

engine 下创建消费者监听器 TestConsumerListener.class

package com.bi.cloud.service.Impl;

import com.alibaba.fastjson.JSON;
import com.bi.cloud.pojo.MyMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 消费者监听器
 **/
@Slf4j
@Component
// topic需要和生产者的topic一致，consumerGroup属性是必须指定的，内容可以随意
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "consumer-group")
public class TestConsumerListener implements RocketMQListener<MyMessage> {

    /**
     * 监听到消息的时候就会调用该方法
     */
    @Override
    public void onMessage(MyMessage message) {
        log.info("从test-topic中监听到消息");
        log.info(JSON.toJSONString(message));
    }
}