CassandraAppender - distributed logging，分布式软件logback-appender

2024-04-09 04:38
文章标签 logging 软件 分布式 logback distributed appender cassandraappender

本文主要是介绍CassandraAppender - distributed logging，分布式软件logback-appender，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

   农历年最后一场scala-meetup听刘颖分享专业软件开发经验，大受启发。突然意识到一直以来都没有完全按照任何标准的开发规范做事。诚然，在做技术调研和学习的过程中不会对规范操作有什么严格要求，一旦技术落地进入应用阶段，开始进行产品开发时，只有严格按照专业的软件开发规范才能保证软件产品的质量。刘颖在meetup中提到了异常处理(exception handling)和过程跟踪（logging)作为软件开发规范中的重要环节。我们在这篇先讨论logging。logging通过记录软件运行过程帮助开发者跟踪软件运行情况，分析运算结果或者异常产生原因，是一个成功完整的软件不可缺少的环节。 logback应该是java生态链中最流行、最通用的logger了。虽然logback已经提供了STDOUT、FILE、DB等多种跟踪信息输出方式，即ConsoleAppender、FileAppender、DBAppender，但针对分布式应用的appender还是需要定制。因为分布式软件是跨系统运行的，跟踪信息自然也会在不同的系统中产生并存储，所以分布式应用需要分布式存储才能实现跟踪信息的全局管理。logback是一套开发架构，任何定制的appender可以很方便地整合入logback。那么我们就尝试开发一套基于cassandra的logback-appender。

首先认识一下logback：感觉需要重点了解的logging运作核心应该是消息等级level的操作。消息等级是指logback根据不同的消息等级来筛选需要记录的消息。logback支持下面几个消息等级，按照各自记录动作覆盖面由弱到强排列，包括：

TRACE -> DEBUG -> INFO -> WARN -> ERROR 分别对应记录函数 trace(msg),debug(msg),info(msg),warn(msg),error(msg)

logback按消息等级进行记录筛选的规则如下：

假设记录函数为p，某个class的消息等级level为q：当p>=q时选择记录消息。换言之调用函数error(msg)时logback会记录所有等级消息，反之trace(msg)只能记录TRACE级别的消息。logback手册中如下表示：

                 TRACE    DEBUG   INFO    WARN    ERROR   OFF
trace()         YES          NO      NO         NO         NO        NO
debug()       YES         YES      NO         NO         NO        NO
info()           YES         YES     YES         NO         NO        NO
warn()         YES         YES     YES        YES         NO        NO
error()         YES         YES     YES        YES        YES        NO

logback中每个类的默认消息等级可以按照类型继承树结构继承。当一个子类没有定义消息等级时，它继承对上父类的消息等级，即：X.Y.Z中Z的默认消息等级从Y继承。

好了，以上运作流程都包括在logback的功能里了，跟消息的存储appender没什么关系。下面我们就开始自制一套基于cassandra的appender。上面提过，logback是一套开放的框架，任何按照logback要求开发的appender都可以很方便的整合入logback的功能中去。下面是一个logback的appender框架：

package com.datatech.logback
import ch.qos.logback.classic.spi.ILoggingEvent
import ch.qos.logback.core.UnsynchronizedAppenderBase
import com.datastax.driver.core.querybuilder.QueryBuilderclass CassandraAppender extends UnsynchronizedAppenderBase[ILoggingEvent]{override def append(eventObject: ILoggingEvent): Unit = {//write log message to cassandra}override def start(): Unit = {//setup cassandrasuper.start()}override def stop(): Unit = {super.stop()//clean up, closing cassandra}}

我们先实现一个完整的logback配置文件logback.xml，包括ConsoleAppender,FileAppender,CassandraAppender 

<configuration><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><encoder><Pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</Pattern></encoder></appender><appender name="FILE" class="ch.qos.logback.core.FileAppender"><!-- path to your log file, where you want to store logs --><file>/Users/Tiger/logback.log</file><append>false</append><encoder><pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern></encoder></appender><appender name="cassandraLogger" class="com.datatech.logback.CassandraAppender"><hosts>192.168.0.189</hosts><port>9042</port><appName>posware</appName><defaultFieldValues>{"app_customer":"bayakala.com","app_device":"1001"}</defaultFieldValues><keyspaceName>applog</keyspaceName><columnFamily>txnlog</columnFamily></appender><root level="debug"><appender-ref ref="cassandraLogger" /><appender-ref ref="STDOUT" /></root><shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
</configuration>

配置文件里CassandraAppender的属性，如hosts,port,keyspaceName等在scala程序实现方法如下：

 private var _hosts: String = ""def setHosts(hosts: String): Unit = _hosts = hostsprivate var _port: Int = 9042 // for the binary protocol, 9160 is default for thriftdef setPort(port: Int): Unit = _port = portprivate var _username: String = ""def setUsername(username: String): Unit = _username = usernameprivate var _password: String = ""def setPassword(password: String): Unit = _password = password

属性的使用如下：

       writeLog(eventObject)(optSession.get, _keyspaceName, _columnFamily)(_appName,ip,hostname,_defaultFieldValues)

实际上logback.xml里的这些属性可以在runtime时设定，如下：

//get appender instancesval log: Logger = LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME).asInstanceOf[Logger]val cassAppender = log.getAppender("cassandraLogger").asInstanceOf[CassandraAppender]val stdoutAppender = log.getAppender("STDOUT").asInstanceOf[ConsoleAppender[ILoggingEvent]]val fileAppender = log.getAppender("FILE").asInstanceOf[FileAppender[ILoggingEvent]]if(cassAppender != null) {cassAppender.setHosts("192.168.0.189")cassAppender.setPort(9042)cassAppender.start()}

与通用的appender不同的是：我们需要在应用中与logback进行互动，因为我们需要把具体应用中一些特定的跟踪目标当作数据库字段记录起来。这些特定的跟踪目标如userid,productid等是应用业务特有的，通用的logger是无法覆盖的。所以我们关注的是一套在应用层面通用的logger。为了实现这一目标，首先可以在数据库表结构schema里表现应用的业务特点，下面是个例子：

CREATE TABLE IF NOT EXISTS applog.txnlog (class_name text,file_name text,host_ip text,host_name text,level text,line_number text,logger_name text,method_name text,thread_name text,throwable_str_rep text,log_date text,log_time text,log_msg text,app_name text,app_customer text,app_device text,PRIMARY KEY (app_customer, app_device, log_date, log_time)
);

以上的schema中app_customer,app_device属于应用业务属性，因为我们希望从用户或设备角度对消息进行分类管理。以此类推对其它应用我们也是通过设计另一套涵盖业务特性的schema。这些反应业务特性的字段必须在应用中调用消息记录函数时提供，因为这些字段的内容是动态的（如：一个服务端软件的用户可能有几百上千个）。我们只能通过记录的消息来传递这些字段的值。记住，logback可以同时支持自带的appender如ConsoleAppender,FileAppender等，以及CassandraAppender,大家共用logback获取的msg，但我们又必须通过对msg的处理才能加入动态属性的值。为了不影响msg的可读性，可以用json来处理msg,如下：

   var msg = event.getMessage()try {val logMap = fromJson[Map[String,String]](msg)logMap.foreach ( m => qryInsert = qryInsert.value(m._1, m._2))} catch {case e: Throwable =>qryInsert = qryInsert.value(MESSAGE, msg)try {val dftMap = fromJson[Map[String,String]](default)dftMap.foreach ( m => qryInsert = qryInsert.value(m._1, m._2))} catch {case e: Throwable => }}session.execute(qryInsert)

如果event.getMessage()获取的msg不是json格式（如：消息是应用中引用的第三方工具库产生的），就采用在配置文件中定义的默认值（也是json格式的），如上面配置文件中的<defaultFieldValues>属性。

cassandra的使用比较简单，而且我们只使用了insert一项操作。完整的CassandraAppender源代码如下：

package com.datatech.logback
import ch.qos.logback.classic.spi._
import ch.qos.logback.core.UnsynchronizedAppenderBase
import com.datastax.driver.core._
import com.datastax.driver.core.querybuilder.{Insert, QueryBuilder}
import java.net.InetAddress
import java.time._
import java.time.format._
import java.util.Localeclass CassandraAppender extends UnsynchronizedAppenderBase[ILoggingEvent]{
import CassandraAppender._private var _hosts: String = ""def setHosts(hosts: String): Unit = _hosts = hostsprivate var _port: Int = 9042 // for the binary protocol, 9160 is default for thriftdef setPort(port: Int): Unit = _port = portprivate var _username: String = ""def setUsername(username: String): Unit = _username = usernameprivate var _password: String = ""def setPassword(password: String): Unit = _password = passwordprivate var _defaultFieldValues: String = ""def setDefaultFieldValues(defaultFieldValues: String) = _defaultFieldValues = defaultFieldValuesprivate val ip: String = getIP()private val hostname: String = getHostName()// Keyspace/ColumnFamily informationprivate var _keyspaceName: String = "Logging"def setKeyspaceName(keyspaceName: String): Unit = _keyspaceName = keyspaceNameprivate var _columnFamily: String = "log_entries"def setColumnFamily(columnFamily: String): Unit = _columnFamily = columnFamilyprivate var _appName: String = "default"def setAppName(appName: String): Unit = _appName = appNameprivate var _replication: String = "{ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }"def setReplication(replication: String): Unit = _replication = replicationprivate var _consistencyLevelWrite: ConsistencyLevel = ConsistencyLevel.ONEdef setConsistencyLevelWrite(consistencyLevelWrite: String): Unit = {try {_consistencyLevelWrite = ConsistencyLevel.valueOf(consistencyLevelWrite.trim)} catch { case e: Throwable =>throw new IllegalArgumentException("Consistency level " + consistencyLevelWrite + " wasn't found.")}}private var optCluster: Option[Cluster] = Noneprivate var optSession: Option[Session] = Nonedef connectDB(): Unit = {try {val cluster = new Cluster.Builder().addContactPoints(_hosts).withPort(_port).build()val session = cluster.connect()optCluster = Some(cluster)optSession = Some(session)} catch {case e: Throwable =>optCluster = NoneoptSession = Noneprintln(s"error when logger connecting to cassandra [${_hosts}:${_port}]")}}override def append(eventObject: ILoggingEvent): Unit = {if(optSession.isDefined) {try {writeLog(eventObject)(optSession.get, _keyspaceName, _columnFamily)(_appName,ip,hostname,_defaultFieldValues)} catch {case e: Throwable =>}}}override def start(): Unit = {if(! _hosts.isEmpty) {connectDB()super.start()}}override def stop(): Unit = {super.stop()if(optSession.isDefined) {optSession.get.closeAsync()optCluster.get.closeAsync()}}}object CassandraAppender extends JsonConverter {// CF column namesval HOST_IP: String = "host_ip"val HOST_NAME: String  = "host_name"val APP_NAME: String = "app_name"val LOGGER_NAME: String = "logger_name"val LEVEL: String  = "level"val CLASS_NAME: String = "class_name"val FILE_NAME: String = "file_name"val LINE_NUMBER: String = "line_number"val METHOD_NAME: String = "method_name"val THREAD_NAME: String = "thread_name"val THROWABLE_STR: String = "throwable_str_rep"val LOG_DATE: String = "log_date"val LOG_TIME: String = "log_time"val MESSAGE: String = "log_msg"val  dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS", Locale.US)def logDate: String = java.time.LocalDate.now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))def logTime: String = LocalDateTime.now.format(dateTimeFormatter).substring(11)def writeLog(event: ILoggingEvent)(session: Session, kspc: String, tbl: String)(appName: String, ip: String, hostName: String, default: String): ResultSet = {var qryInsert = QueryBuilder.insertInto(kspc,tbl).value(APP_NAME,appName).value(HOST_IP,ip).value(HOST_NAME,hostName).value(LOGGER_NAME,event.getLoggerName()).value(LEVEL,event.getLevel().toString).value(THREAD_NAME,event.getThreadName()).value(LOG_DATE,logDate).value(LOG_TIME,logTime)try {val callerData = event.getCallerData()if (callerData.nonEmpty) {qryInsert = qryInsert.value(CLASS_NAME, callerData.head.getClassName()).value(FILE_NAME, callerData.head.getFileName()).value(LINE_NUMBER, callerData.head.getLineNumber().toString).value(METHOD_NAME, callerData.head.getMethodName())}} catch {case e: Throwable => println(s"logging event error: ${e.getMessage}")}try {if (event.getThrowableProxy() != null) {val throwableStrs = event.getThrowableProxy().getSuppressed().asInstanceOf[List[IThrowableProxy]]val throwableStr = throwableStrs.foldLeft("") { case (b, t) => b + "," + t.getMessage() }qryInsert = qryInsert.value(THROWABLE_STR, throwableStr)}} catch {case e: Throwable => println(s"logging event error: ${e.getMessage}")}var msg = event.getMessage()try {val logMap = fromJson[Map[String,String]](msg)logMap.foreach ( m => qryInsert = qryInsert.value(m._1, m._2))} catch {case e: Throwable =>qryInsert = qryInsert.value(MESSAGE, msg)try {val dftMap = fromJson[Map[String,String]](default)dftMap.foreach ( m => qryInsert = qryInsert.value(m._1, m._2))} catch {case e: Throwable => }}session.execute(qryInsert)}def getHostName(): String = {var hostname = "unknown"try {val addr: InetAddress = InetAddress.getLocalHost()hostname = addr.getHostName()} catch { case e: Throwable => hostname = "error"}hostname}def getIP(): String = {var ip: String = "unknown"try {val addr: InetAddress = InetAddress.getLocalHost()ip = addr.getHostAddress()} catch { case e: Throwable => ip = "error" }ip}}

下面是测试代码：

import ch.qos.logback.classic.Logger
import ch.qos.logback.core.{ConsoleAppender, FileAppender}
import com.datatech.logback.{CassandraAppender, JsonConverter}
import ch.qos.logback.classic.spi.ILoggingEvent
import org.slf4j.LoggerFactory
import ch.qos.logback.classic.LoggerContext
import java.time._
import java.time.format._
import java.util.Localeimport scala.io._
import com.datastax.driver.core._object LoggingDemo extends App with JsonConverter {val log: Logger = LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME).asInstanceOf[Logger]val cassAppender = log.getAppender("cassandraLogger").asInstanceOf[CassandraAppender]val stdoutAppender = log.getAppender("STDOUT").asInstanceOf[ConsoleAppender[ILoggingEvent]]val fileAppender = log.getAppender("FILE").asInstanceOf[FileAppender[ILoggingEvent]]/*val cluster = new Cluster.Builder().addContactPoints("192.168.0.189").withPort(9042).build()val session = cluster.connect()val keyspace = getClass.getResource("/logger.schema")val table = getClass.getResource("/txnlog.schema")val qrykspc = Source.fromFile(keyspace.getPath).getLines.mkStringsession.execute(qrykspc)val qrytbl = Source.fromFile(table.getPath).getLines.mkStringsession.execute(qrytbl)session.close()cluster.close()val json = toJson(loggedItems)println(s"json = $json")val m = fromJson[Map[String,String]](json)println(s"map = $m")//stop the appendersif (stdoutAppender != null)stdoutAppender.stop()if (fileAppender != null)fileAppender.stop()
*/if(cassAppender != null) {cassAppender.setHosts("192.168.0.189")cassAppender.setPort(9042)cassAppender.start()}val  dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS", Locale.US)val now = LocalDateTime.now.format(dateTimeFormatter)log.info("************this is a info message ..." + now)log.debug("***********debugging message here ..." + now)var loggedItems = Map[String,String]()
//  loggedItems += ("app_name" -> "test")loggedItems = loggedItems ++ Map(("app_customer" -> "logback.com"),("app_device" -> "9101"),("log_msg" -> "specific message for cassandra ..."))log.debug(toJson(loggedItems))//stop the loggerval loggerContext = LoggerFactory.getILoggerFactory.asInstanceOf[LoggerContext]loggerContext.stop()
}

 

 

 

 

 

 

 

 

这篇关于CassandraAppender - distributed logging，分布式软件logback-appender的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

原文地址:
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.chinasem.cn/article/887195。

相关文章

SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志

SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志

《SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志》在SpringBoot项目中,使用logback-spring.xml配置屏蔽特定路径的日志有两种常用方式,文中的... 目录方案一：基础配置（直接关闭目标路径日志）方案二：结合 Spring Profile 按环境屏蔽关
阅读更多...
安装centos8设置基础软件仓库时出错的解决方案

安装centos8设置基础软件仓库时出错的解决方案

《安装centos8设置基础软件仓库时出错的解决方案》：本文主要介绍安装centos8设置基础软件仓库时出错的解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录安装Centos8设置基础软件仓库时出错版本 8版本 8.2.200android4版本 javas
阅读更多...
如何确定哪些软件是Mac系统自带的? Mac系统内置应用查看技巧

如何确定哪些软件是Mac系统自带的? Mac系统内置应用查看技巧

《如何确定哪些软件是Mac系统自带的?Mac系统内置应用查看技巧》如何确定哪些软件是Mac系统自带的？mac系统中有很多自带的应用，想要看看哪些是系统自带，该怎么查看呢？下面我们就来看看Mac系统内... 在MAC电脑上，可以使用以下方法来确定哪些软件是系统自带的：1.应用程序文件夹打开应用程序文件夹
阅读更多...
Logback在SpringBoot中的详细配置教程

Logback在SpringBoot中的详细配置教程

《Logback在SpringBoot中的详细配置教程》SpringBoot默认会加载classpath下的logback-spring.xml（推荐）或logback.xml作为Logback的配置... 目录1. Logback 配置文件2. 基础配置示例3. 关键配置项说明Appender（日志输出器
阅读更多...
Golang实现Redis分布式锁(Lua脚本+可重入+自动续期)

Golang实现Redis分布式锁(Lua脚本+可重入+自动续期)

《Golang实现Redis分布式锁(Lua脚本+可重入+自动续期)》本文主要介绍了Golang分布式锁实现,采用Redis+Lua脚本确保原子性,持可重入和自动续期,用于防止超卖及重复下单,具有一定... 目录1 概念应用场景分布式锁必备特性2 思路分析宕机与过期防止误删keyLua保证原子性可重入锁自动
阅读更多...
基于MongoDB实现文件的分布式存储

基于MongoDB实现文件的分布式存储

《基于MongoDB实现文件的分布式存储》分布式文件存储的方案有很多,今天分享一个基于mongodb数据库来实现文件的存储,mongodb支持分布式部署,以此来实现文件的分布式存储,需要的朋友可以参考... 目录一、引言二、GridFS 原理剖析三、Spring Boot 集成 GridFS3.1 添加依赖
阅读更多...
Python logging模块使用示例详解

Python logging模块使用示例详解

《Pythonlogging模块使用示例详解》Python的logging模块是一个灵活且强大的日志记录工具,广泛应用于应用程序的调试、运行监控和问题排查,下面给大家介绍Pythonlogging模... 目录一、为什么使用 logging 模块？二、核心组件三、日志级别四、基本使用步骤五、快速配置（bas
阅读更多...
Spring Boot集成Logback终极指南之从基础到高级配置实战指南

Spring Boot集成Logback终极指南之从基础到高级配置实战指南

《SpringBoot集成Logback终极指南之从基础到高级配置实战指南》Logback是一个可靠、通用且快速的Java日志框架,作为Log4j的继承者,由Log4j创始人设计,：本文主要介绍... 目录一、Logback简介与Spring Boot集成基础1.1 Logback是什么？1.2 Sprin
阅读更多...
Redis实现分布式锁全解析之从原理到实践过程

Redis实现分布式锁全解析之从原理到实践过程

《Redis实现分布式锁全解析之从原理到实践过程》：本文主要介绍Redis实现分布式锁全解析之从原理到实践过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、背景介绍二、解决方案（一）使用 SETNX 命令（二）设置锁的过期时间（三）解决锁的误删问题（四）Re
阅读更多...
Gradle下如何搭建SpringCloud分布式环境

Gradle下如何搭建SpringCloud分布式环境

《Gradle下如何搭建SpringCloud分布式环境》：本文主要介绍Gradle下如何搭建SpringCloud分布式环境问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地... 目录Gradle下搭建SpringCloud分布式环境1.idea配置好gradle2.创建一个空的gr
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2