@

目录

• StringBoot整合ELK实现日志收集和搜索自动补全功能(详细图文教程)


• 一、下载ELK的安装包上传并解压
  
  
  
  • 1.Elasticsearch下载
  
  
  • 2.Logstash下载
  
  
  • 3.Kibana下载
  
  
  • 4.上传并解压
  
  
  
  


• 二、Elasticsearch配置
  
  
  
  • 1.修改配置文件elasticsearch.yml
  
  
  • 2.新建es用户
  
  
  • 3.启动Elasticsearch
  
  
  • 4.启动问题：
    
    
    
    • 4.1：JDK版本问题：
    
    
    
    
  
  
  
  


• 配置自己的jdk11


• 添加jdk判断
  
  
  
  • 
    
    
    
    • 4.2：内存不足问题处理：
    
    
    
    
  
  
  
  


• 三、Logstash配置
  
  
  
  • 1.新建持久化目录：
  
  
  • 2.修改配置文件logstash.yml
  
  
  • 3.安装logstash所需插件
  
  
  • 4.新建配置文件 logstash.conf（/bin目录下）
  
  
  • 5.启动logstash
  
  
  • 6.启动问题：
    
    
    
    • 6.1：内存问题：
    
    
    
    
  
  
  
  


• 四、Kibana配置
  
  
  
  • 1.修改配置文件kibana.yml
  
  
  • 2.启动kibana
  
  
  
  


• 五、Spring boot整合ELK实现日志采集
  
  
  
  • 1. 修改 pom.xml
  
  
  • 2. logback.xml
  
  
  • 3. Test.java
  
  
  • 4. 进入kibana -> 管理 -> 索引模式 -> 创建索引模式,输入 * ，下一步创建
  
  
  • 5. 调用接口触发日志记录，打开Kibana-->Discover菜单-->筛选" * ",显示请求日志
  
  
  
  


• 六、使用logstash-input-jdbc进行Mysql与ES数据同步
  
  
  
  • 1，下载 mysql-connector-java-8.0.20.jar 包放到安装目录，复制绝对路径下面会用到
  
  
  • 2，新建配置jdbc.conf和jdbc.sql文件
  
  
  
  


• 过滤处理,如果需要，也可参考elk安装那篇
  
  
  
  • 3，然后到bin目录下启动 Logstash
  
  
  
  


• 七、使用ES实现简单的搜索自动补全功能
  
  
  
  • 1，安装es客户端pom.xml依赖包
  
  
  • 2，编写ES-Java客户端调用代码
  
  
  • 3,Service
  
  
  • 4,Controller
  
  
  • 前端代码vue
  
  
  
  

StringBoot整合ELK实现日志收集和搜索自动补全功能(详细图文教程)

最近在给小组搭建ELK环境，并集成到项目中使用，这里记录下（jdk等基础环境配置请自行百度）

---

一、下载ELK的安装包上传并解压

我这边用到的版本是7.6.1版本，ELK最好版本统一，避免版本兼容导致失败,可以在官网下载，下载可能会慢，我这里下载好了，可以直接用

1.Elasticsearch下载

elasticsearch-7.6.1-linux-x86_64.tar.gz

链接：https://pan.baidu.com/s/1cAswtaYx8_2Q56XPZN_ZJA

提取码：idn3

2.Logstash下载

logstash-7.6.1.tar.gz

链接：https://pan.baidu.com/s/1bXAMXSVEe1zpOx3xIbF8LA

提取码：475x

3.Kibana下载

kibana-7.6.1-linux-x86_64.tar.gz

链接：https://pan.baidu.com/s/1xTxTLEooqCHypMIFB_tPng

提取码：w6ob

上传到自定义目录：

/usr/local/src/software


4.上传并解压

接下来使用 XShell 工具连接到服务器或者虚拟机(我使用的是阿里云服务器)

ES压缩包解压：进入 /usr/local/src/software 目录

elasticsearch解压命令：tar -zxvf elasticsearch-7.6.1-linux-x86_64.tar.gz

logstash解压命令：tar -zxvf logstash-7.6.1.tar.gz

kibana解压命令：tar -zxvf kibana-7.6.1-linux-x86_64.tar.gz

解压后的文件夹修改下名字，方便输入：mv elasticsearch-7.6.1 elasticsearch7

解压后的文件夹修改下名字，方便输入：mv kibana-7.6.1 kibana7

ES目录介绍

bin:下面存放着Es启动文件 elasticsearch.bat/elasticsearch
config：配置目录
data:数据目录
jdk、lib：Java运行环境以及依赖包
logs:日志目录
modules、plugins：模块及插件目录，head插件可以存放在plugins目录下


二、Elasticsearch配置

1.修改配置文件elasticsearch.yml

基础配置

ES 本身其实也相当于是一个数据库，为此，我们在 elasticsearch7 文件夹下自己建一个 data 文件夹，用于存放数据：

mdkir data


进入 data 文件夹下，执行 pwd 命令拷贝下该文件夹的路径，下面配置要用到！

我的 data 路径：

/usr/local/src/software/elasticsearch7/data


修改配置文件elasticsearch.yml，我们进入elasticsearch7 这个文件夹下的config 文件夹，编辑 elasticsearch.yml 文件：

vim elasticsearch.yml


跨域问题

http.cors.enabled: true
http.cors.allow-origin: "*"
http.cors.allow-headers: Authorization,X-Requested-With,Content-Length,Content-Type


保存退出esc+(:wq!)

2.新建es用户

因为Elasticsearch为了安全起见，不支持root用户直接启动

新建一个用户elasticsearch，命令：

adduser es


//添加一个名为tommy的用户
#passwd tommy //修改密码
Changing password for user tommy.
New UNIX password: //在这里输入新密码
Retype new UNIX password: //再次输入新密码
passwd: all authentication tokens updated successfully.


在software 目录下 赋予 elasticsearch7 这个文件夹的权限给 es用户，命令：

chown -R es:es /usr/local/src/software/elasticsearch7


命令，切换到 es 用户，并重新到 bin 目录下执行 ES服务

su es


3.启动Elasticsearch

在es安装目录bin下执行

./elasticsearch -d


浏览器访问：http://服务器ip:9200/

显示如下内容表示成功

4.启动问题：

由于某种原因可能导致启动失败，我这边也遇到过一些情况，可以参考解决方案：

4.1：JDK版本问题：

future versions of Elasticsearch will require Java 11; your Java version from [/usr/local/jdk8/jdk1.8.0_291/jre] does not meet this requirement


原因：说明这个版本对应的jdk应该是java11。我们环境中的jdk是java8。其实新版的es是自带了jdk的。但是和我们环境变量的冲突了。

解决方案：修改启动文件,添加如下修改

wget https://download.java.net/java/GA/jdk11/13/GPL/openjdk-11.0.1_linux-x64_bin.tar.gz
tar -xzvf openjdk-11.0.1_linux-x64_bin.tar.gz /opt/


vi bin/elasticsearch


配置自己的jdk11

export JAVA_HOME=/opt/jdk-11.0.1
export PATH=$JAVA_HOME/bin:$PATH


添加jdk判断

if [ -x "$JAVA_HOME/bin/java" ]; then
JAVA="/opt/jdk-11.0.1/bin/java"
else
JAVA=`which java`
fi


4.2：内存不足问题处理：

可以修改 config 下的 jvm.options 配置文件，将运行大小 2g 修改为 1g(还是不行的话，再小设置到256m)：

三、Logstash配置

1.新建持久化目录：

mkdir -p /usr/local/src/software/logstash-7.6.1/plugin-data


2.修改配置文件logstash.yml

编辑 /config/logstash.yml 配置文件

node.name: cpy04.dev.xjh.com #设置节点名称，一般写主机名
path.data: /usr/local/src/software/logstash-7.6.1/plugin-data #上一步创建logstash 和插件使用的持久化目录
config.reload.automatic: true #开启配置文件自动加载
config.reload.interval: 10 #定义配置文件重载时间周期
http.host: "172.18.10.172" #(阿里云服务器私网IP)定义访问主机名，一般为域名或IP


3.安装logstash所需插件

从Filebeat 输入、过滤、输出至elasticsearch（logstash 有非常多插件，详见官网,此处不列举）

安装logstash-input-jdbc 和logstash-input-beats-master 插件

如果你的Logstash没有安装logstash-codec-json_lines插件，通过以下命令安装

/usr/local/src/software/logstash-7.6.1/bin/logstash-plugin install logstash-integration-jdbc
wget https://github.com/logstash-plugins/logstash-input-beats/archive/master.zip -O /opt/master.zip
unzip -d /usr/local/src/software/logstash-7.6.1 /opt/master.zip


root@iZwz91w9jegcgf28ttbe2yZ:/usr/local/src/software/logstash-7.6.1/bin# ./logstash-plugin install logstash-codec-json_lines
Validating logstash-codec-json_lines
Installing logstash-codec-json_lines
Installation successful


4.新建配置文件 logstash.conf（/bin目录下）

vim logstash.conf


配置文档内容

input {
tcp {
#模式选择为server
mode => "server"
#阿里云服务器私网ip和端口根据自己情况填写，端口默认4560,我这边改成4567，对应后面整合spring boot的logback.xml里appender中的destination
host => "172.18.10.172"
port => 4567
#格式json
codec => json_lines
}
}
filter {
#过滤器，根据需要填写
}
output {
elasticsearch {
action => "index"
#这里是es的地址（es服务在阿里云服务器私网ip），多个es要写成数组的形式
hosts => "172.18.10.172:9200"
#用于kibana过滤，可以填项目名称
index => "springboot-logstash-%{+YYYY.MM.dd}"
}
}


5.启动logstash

切换到/usr/local/src/software/logstash-7.6.1/bin

nohup ./logstash -f logstash.conf >/dev/null 2>&1 &


验证

访问http://服务器ip:9600/，成功的话会显示一个JSON串

6.启动问题：

由于某种原因可能导致启动失败，我这边也遇到过一些情况，可以参考解决方案：

6.1：内存问题：

:Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c5330000


解决方案;

.修改logstash核心的执行文件 在bin目录下，有一个可执行的文件logstash,需要添加参数：（可以先

不用设置，报错的话再设置）

LS_JAVA_OPTS="-server -Xms256m -Xmx512m -XX:PermSize=128m -XX:MaxPermSize=256m"


2./usr/local/src/software/logstash-7.6.1/config修改jvm.options

-Xms256m
-Xmx256m


四、Kibana配置

1.修改配置文件kibana.yml

修改内容：

2.启动kibana

进入bin目录启动: ./bin/kibana
或者守护进程启动: nohup ./bin/kibana &
或者 nohup ./bin/kibana >/dev/null 2>&1 &

关键在于最后的 >/dev/null 2>&1 部分，/dev/null是一个虚拟的空设备（类似物理中的黑洞），任何输出信息被重定向到该设备后，将会石沉大海
>/dev/null 表示将标准输出信息重定向到"黑洞"
2>&1 表示将标准错误重定向到标准输出(由于标准输出已经定向到“黑洞”了，即：标准输出此时也是"黑洞"，再将标准错误输出定向到标准输出，相当于错误输出也被定向至“黑洞”)


进行访问本机访问 http://服务器ip:5601

五、Spring boot整合ELK实现日志采集

日志采集新增Logback直接发送日志到Logstash的形式。如果采用此方式，web服务可减少部分生成log文件配置，提高实时性和日志推送效率

1. 修改 pom.xml

logback 相关



ch.qos.logback
logback-classic


ch.qos.logback
logback-core


ch.qos.logback
logback-classic


ch.qos.logback
logback-access


net.logstash.logback
logstash-logback-encoder
5.3



2. logback.xml













${APP_NAME}



${LOG_FILE_PATH}/${APP_NAME}-%d{yyyy-MM-dd}.log
30




${FILE_LOG_PATTERN}





172.18.10.172:4567










3. Test.java

在接口中使用log进行日志生成，启动项目

4. 进入kibana -> 管理 -> 索引模式 -> 创建索引模式,输入 * ，下一步创建

5. 调用接口触发日志记录，打开Kibana-->Discover菜单-->筛选" * ",显示请求日志

六、使用logstash-input-jdbc进行Mysql与ES数据同步

1，下载 mysql-connector-java-8.0.20.jar 包放到安装目录，复制绝对路径下面会用到

2，新建配置jdbc.conf和jdbc.sql文件

在bin目录下创建，

jdbc.sql文件

select
*
from user


jdbc.conf 配置文件内容如下

input {
stdin {
}
jdbc {
# mysql相关jdbc配置
jdbc_connection_string => "jdbc:mysql://mysqlip:3306/vueblog2"
jdbc_user => "root"
jdbc_password => "*****"

# jdbc连接mysql驱动的文件目录，可去官网下载:https:#dev.mysql.com/downloads/connector/j/
jdbc_driver_library => "/usr/local/src/software/logstash-7.6.1/mysql-connector-java-8.0.20.jar"
# the name of the driver class for mysql
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
# mysql文件, 也可以直接写SQL语句在此处，如下：
# statement => "SELECT * from user;"
statement_filepath => "/usr/local/src/software/logstash-7.6.1/bin/jdbc.sql"
# 这里类似crontab,定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新，比如每10分钟("*/10 * * * *")执行一次同步
schedule => "* * * * *"
#设定ES索引类型
type => "jdbc"
# 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
#record_last_run => "true"
# 是否需要记录某个column 的值,如果record_last_run为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
#use_column_value => "true"
# 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是mysql主键
#tracking_column => "id"
#last_run_metadata_path => "/opt/logstash/conf/last_id"
# 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
#clean_run => "false"
#是否将 字段(column) 名称转小写
#lowercase_column_names => "false"
}
}


过滤处理,如果需要，也可参考elk安装那篇

filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {
# 输出到elasticsearch的配置
elasticsearch {
hosts => ["http://ESIP:9200/"]
index => "jdbc"
document_type => "user"
#自增ID编号
document_id => "%{id}"
template_overwrite => true
}
# 这里输出调试，正式运行时可以注释掉
stdout {
codec => json_lines
}
}


在logstash安装目录bin下面创建文件夹 logstash_jdbc，将之前的 logstash.conf和刚才创建的jdbc.conf 移动到 logstash_jdbc文件夹下。后面启动logstash会用到

3，然后到bin目录下启动 Logstash

nohup ./logstash -f logstash_jdbc/ >/dev/null 2>&1 &


七、使用ES实现简单的搜索自动补全功能

1，安装es客户端pom.xml依赖包



org.elasticsearch.client
elasticsearch-rest-high-level-client
7.6.1


org.elasticsearch
elasticsearch


org.elasticsearch.client
elasticsearch-rest-client




org.elasticsearch.client
elasticsearch-rest-client
7.6.1


org.elasticsearch
elasticsearch
7.6.1



2，编写ES-Java客户端调用代码

/**
* Created by fansongsong
*/
@Component
@Service
public class Util {
public static User getCurrentUser() {
User user = (User) SecurityContextHolder.getContext().getAuthentication().getPrincipal();
return user;
}
private static RestHighLevelClient client;
public Util() {
RestHighLevelClient client = new RestHighLevelClient(
RestClient.builder(
new HttpHost("ESIP", 9200, "http")));
this.client = client;
}
public void shutdown(){
if(client!=null){
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 根据指定的内容，查询所有Doc。
* @return
*/
public static List> searchArticle(String index,String key, String value){
SearchHits searchHits = search(index,key,value);
List> list = new ArrayList<>();
for(SearchHit hit:searchHits.getHits()){
System.out.println( hit.getSourceAsString());
Map stringObjectMap = hit.getSourceAsMap();
stringObjectMap.put("name", stringObjectMap.get("nickname"));
list.add(stringObjectMap);
}
return list;
}
public static SearchHits search(String index, String key, String value){
QueryBuilder matchQueryBuilder = QueryBuilders.matchPhraseQuery(key, value);
// matchQueryBuilder.
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
// sourceBuilder.query(QueryBuilders.termQuery("content", content));
sourceBuilder.query(matchQueryBuilder);
sourceBuilder.from(0);
sourceBuilder.size(100);
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(index);
searchRequest.source(sourceBuilder);
SearchResponse searchResponse;
List> list = new ArrayList<>();
SearchHits searchHits = null;
try {
searchRespOnse= client.search(searchRequest,RequestOptions.DEFAULT);
searchHits = searchResponse.getHits();
for(SearchHit hit:searchHits.getHits()){
System.out.println( hit.getSourceAsString());
list.add(hit.getSourceAsMap());
}
} catch (IOException e) {
e.printStackTrace();
}
return searchHits;
}
}


3,Service

public List getUserByNicknameKeyword(String nickname) {
List> mapList = Util.searchArticle("jdbc", "nickname", nickname);
List goodsNameList = mapList.stream().map(m->m.get("nickname").toString()).collect(Collectors.toList());
log.info(nickname+"==>搜索自动补全:"+JSONArray.toJSONString(mapList));
return goodsNameList;
}


4,Controller

@RestController
@RequestMapping("/admin")
public class UserManaController {
@Autowired
UserService userService;

@RequestMapping(value = "/user/keyword", method = RequestMethod.GET)
@ResponseBody
public List getUserByNicknameKeyword(String nickname) {
return userService.getUserByNicknameKeyword(nickname);
}
}


前端代码vue


placeholder="默认展示部分用户，可以通过用户名搜索用户..."
prefix-icon="el-icon-search"
v-model="keywords" size="small"
@keyup="get($event)" @keydown.down.prevent="selectDown"
@keydown.up.prevent="selectUp"
>
搜索



• {{value}}

methods: {
get:function (event) {
if(event.keyCode==38||event.keyCode==40)return;
var _this = this;
getRequest("/admin/user/keyword?nickname="+this.keywords).then(resp=> {
_this.loading = false;
if (resp.status == 200) {
_this.myData = resp.data;
} else {
_this.$message({type: 'error', message: '数据加载失败!'});
}
}, resp=> {
_this.loading = false;
if (resp.response.status == 403) {
var data = resp.response.data;
_this.$message({type: 'error', message: data});
}
});
},
selectDown:function () {
this.now++;
if(this.now==this.myData.length)this.now=-1;
this.keywords=this.myData[this.now];
},
selectUp:function () {
this.now--;
if(this.now==-2)this.now=this.myData.length-1;
this.keywords=this.myData[this.now];
}
},
data(){
return {
loading: false,
keywords: '',
myData:[],
now:-1
}
}



显示效果