[大数据]springboot整合消息索引

ElasticSearch检索

1.1、初识ElasticSearch

我们的应用经常需要添加检索功能，开源的 ElasticSearch 是目前全文搜索引擎的首选。他可以快速的存储、搜索和分析海量数据。Spring Boot通过整合SpringData ElasticSearch为我们提供了非常便捷的检索功能支持。

Elasticsearch是一个分布式搜索服务，提供Restful API，底层基于Lucene，采用多shard（分片）的方式保证数据安全，并且提供自动resharding的功能，github等大型的站点也是采用了ElasticSearch作为其搜索服务。

官方原话：

Elasticsearch 是一个分布式的开源搜索和分析引擎，适用于所有类型的数据，包括文本、数字、地理空间、结构化和非结构化数据。Elasticsearch 在 Apache Lucene 的基础上开发而成，由 Elasticsearch N.V.（即现在的 Elastic）于 2010 年首次发布。Elasticsearch 以其简单的 REST 风格 API、分布式特性、速度和可扩展性而闻名，是 Elastic Stack 的核心组件；Elastic Stack 是适用于数据采集、充实、存储、分析和可视化的一组开源工具。人们通常将 Elastic Stack 称为 ELK Stack（代指 Elasticsearch、Logstash 和 Kibana），目前 Elastic Stack 包括一系列丰富的轻量型数据采集代理，这些代理统称为 Beats，可用来向 Elasticsearch 发送数据。

1.2、ElasticSearch基本概念

1、Index (索引)

动词：相当于MySQL中的insert
名词：相当于MySQL中的Database
2、Type (类型)

在 Index (索引) 中，可以定义一个或多个类型，类似于MySQL中的Table，每一种类型的数据放在一起。
3、Document (文档)

保存在某个索引 (index) 下，某种类型 (Type) 的一个数据 (Document)。文档是JSON格式的，Document就像是MySQL中的某个table里面的内容。
4、概念关系图一览

5、倒排索引机制

1.3、ElasticSearch与kibana安装

为了安装方便，我们还是以容器化技术Docker进行安装：

1、下载镜像文件

[howie@laizhenghua /]$ sudo docker search elasticsearch

docker pull elasticsearch:7.4.2 # 存储和检索数据
dicker pull kibana:7.4.2 # 可视化检索数据

[howie@laizhenghua /]$ sudo docker images
REPOSITORY      TAG       IMAGE ID       CREATED         SIZE
mysql           5.7       f07dfa83b528   3 weeks ago     448MB
redis           latest    ef47f3b6dc11   5 weeks ago     104MB
kibana          7.4.2     230d3ded1abc   14 months ago   1.1GB
elasticsearch   7.4.2     b1179d41a7b4   14 months ago   855MB

2、创建挂载目录

sudo mkdir -p /mydata/elasticsearch/config
sudo mkdir -p /mydata/elasticsearch/data
sudo chmod -R 777 /mydata/elasticsearch/
sudo echo "http.host: 0.0.0.0" >/mydata/elasticsearch/config/elasticsearch.yml

chmod：change mode -R(Recursion 递归)

3、启动elasticsearch

sudo docker run --name elasticsearch -p 9200:9200 -p 9300:9300 \
-e  "discovery.type=single-node" \
-e ES_JAVA_OPTS="-Xms64m -Xmx512m" \
-v /mydata/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml \
-v /mydata/elasticsearch/data:/usr/share/elasticsearch/data \
-v  /mydata/elasticsearch/plugins:/usr/share/elasticsearch/plugins \
-d elasticsearch:7.4.2

4、启动 kibana

sudo docker run --name kibana -e ELASTICSEARCH_HOSTS=http://https://es-0a8t6rzt.public.tencentelasticsearch.com:9200/lanlan/Article -p 5601:5601 -d kibana:7.4.2

[howie@laizhenghua /]$ sudo docker run --name kibana -e ELASTICSEARCH_HOSTS=https://es-0a8t6rzt.public.tencentelasticsearch.com:9200/lanlan/Article -p 5601:5601 -d kibana:7.4.2
9b88afc2f1c00891125eeb9610bb149a023b8044bb887568219f7f160059005d

访问服务器的5601端口！

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RpqstGM6-1644044896802)(C:\Users\86157\AppData\Local\Temp\1643814866889.png)]

由于我之前的阿里云服务器的内存不够，于是我申请了一个腾讯云的es。

这次我们用到的是简单的ES的整合。

2.1整合

1.导入依赖

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

2.创建实体类

@Document(indexName = "lanlan")
@Data
public class Article {
    @Id
    private Integer id;
    @Field(type = FieldType.Text)
    private String author;
    @Field(type = FieldType.Text,analyzer = "ik_max_word")
    private String title;
}

3.实现ElasticsearchRepository

public interface ArticleRepository extends ElasticsearchRepository<Article,Integer> {
    List<Article> findArticleByTitle(String title);
    List<Article> findByTitle(String title);
    List<Article> findByTitleContaining(String title);
}

4.测试

@SpringBootTest
class Demo12ElasticsearchApplicationTests {


    @Autowired
    ArticleRepository articleRepository;

    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Test
    public void test03(){
   /*     Iterable<Article> all = articleRepository.findAll();
        for (Article article : all) {
            System.out.println(article);
        }*/


        List<Article> Forest = articleRepository.findByTitleContaining("森林");
        for (Article article : Forest) {
            System.out.println(article);
        }
    }

    @Test
    public void test02(){
        Article article = new Article();
        article.setId(3);
        article.setTitle("挪威森林1");
        article.setAuthor("村上春树");
        articleRepository.save(article);
    }



}

ElasticsearchRepository中的方法名对应我在官网中找到了这样的依据，现在粘贴过来。

如表所示

消息

1.1、前言

在介绍RabbitMQ之前，我们先来看下面一个电商项目的场景：

• 商品的原始数据保存在数据库中，增删改查都在数据库中完成。

• 搜索服务数据来源是索引库（Elasticsearch），如果数据库商品发生变化，索引库数据不能及时更新。

• 商品详情做了页面静态化处理，静态页面数据也不会随着数据库商品更新而变化。

如果我们在后台修改了商品的价格，搜索页面和商品详情页显示的依然是旧的价格，这样显然不对。该如何解决？

我们可能会想到这么做：

• 方案1：每当后台对商品做增删改操作，同时修改索引库数据及更新静态页面。

• 方案2：搜索服务和商品页面静态化服务对外提供操作接口，后台在商品增删改后，调用接口。

这两种方案都有个严重的问题：就是代码耦合，后台服务中需要嵌入搜索和商品页面服务，违背了微服务的独立原则。

这时，我们就会采用另外一种解决办法，那就是消息队列！

商品服务对商品增删改以后，无需去操作索引库和静态页面，只需向MQ发送一条消息（比如包含商品id的消息），也不关心消息被谁接收。 搜索服务和静态页面服务监听MQ，接收消息，然后分别去处理索引库和静态页面（根据商品id去更新索引库和商品详情静态页面）。

什么是消息队列
MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

开发中消息队列通常有如下应用场景：

1、任务异步处理：

高并发环境下，由于来不及同步处理，请求往往会发生堵塞，比如说，大量的insert，update之类的请求同时到达MySQL，直接导致无数的行锁表锁，甚至最后请求会堆积过多，从而触发too many connections错误。通过使用消息队列，我们可以异步处理请求，从而缓解系统的压力。将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。减少了应用程序的响应时间。

2、应用程序解耦合：

MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。

1.2、AMQP和JMS

MQ是消息通信的模型，并发具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

两者间的区别和联系：

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式

JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。

JMS规定了两种消息模型；而AMQP的消息模型更加丰富

1.3、常见MQ产品

• ActiveMQ：基于JMS

• RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好

• RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会-

• Kafka：分布式消息系统，高吞吐量

2、RabbitMQ快速入门

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。RabbitMQ官方地址：http://www.rabbitmq.com

2.1、安装

安装RabbitMQ

docker pull rabbitmq

这里是直接安装最新的，如果需要安装其他版本在rabbitmq后面跟上版本号即可

启动RabbitMQ

docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq

这是我们的转换器

Data\Local\Temp\1644044635591.png)]

这是我们的队列

2.2、springboot整合

使用也是非常简单

首先导入amqp的坐标

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.4.2</version>
</dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

配置文件配置

spring:
  rabbitmq:
    host: 8.142.109.15
    username: guest
    password: guest

启动类

@EnableRabbit
@SpringBootApplication
public class Demo10RabbitMqApplication {

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

}

序列化配置

有了这个bean才能在消息队列中添加类

@Configuration
public class MyAMQPConfig {

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

bean类

@Data
public class Book {
    private String bookName;
    private String author;
}

监听的service

@Service
public class BookService {

    /**
     * 监听队列
     * @param book
     */
    @RabbitListener(queues = "atguigu.news")
    public void receive(Book book){
        System.out.println("收到消息"+book);
    }
}

测试添加消息的类

@Test
void boottest(){
    Book book = new Book();
    book.setBookName("挪威的森林");
    book.setAuthor("村上春书");
    rabbitTemplate.convertAndSend("exchange.direct","atguigu.news",book);
}

控制消息队列测试

    @Test
    public void createEchange(){
//        amqpAdmin.declareExchange(new DirectExchange("xiaolan.direct"));
//        amqpAdmin.declareQueue(new Queue("amqpadmin.queue"));
        amqpAdmin.declareBinding(new Binding("amqpadmin.queue",Binding.DestinationType.QUEUE,"xiaolan.direct","amqp.haha",null));
        System.out.println("创建完成");
    }

控制消息队列测试

    @Test
    public void createEchange(){
//        amqpAdmin.declareExchange(new DirectExchange("xiaolan.direct"));
//        amqpAdmin.declareQueue(new Queue("amqpadmin.queue"));
        amqpAdmin.declareBinding(new Binding("amqpadmin.queue",Binding.DestinationType.QUEUE,"xiaolan.direct","amqp.haha",null));
        System.out.println("创建完成");
    }