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。

相关文章

java如何分布式锁实现和选型

java如何分布式锁实现和选型

《java如何分布式锁实现和选型》文章介绍了分布式锁的重要性以及在分布式系统中常见的问题和需求,它详细阐述了如何使用分布式锁来确保数据的一致性和系统的高可用性,文章还提供了基于数据库、Redis和Zo... 目录引言：分布式锁的重要性与分布式系统中的常见问题和需求分布式锁的重要性分布式系统中常见的问题和需求
阅读更多...
Golang使用etcd构建分布式锁的示例分享

Golang使用etcd构建分布式锁的示例分享

《Golang使用etcd构建分布式锁的示例分享》在本教程中,我们将学习如何使用Go和etcd构建分布式锁系统,分布式锁系统对于管理对分布式系统中共享资源的并发访问至关重要,它有助于维护一致性,防止竞... 目录引言环境准备新建Go项目实现加锁和解锁功能测试分布式锁重构实现失败重试总结引言我们将使用Go作
阅读更多...
Redis分布式锁使用及说明

Redis分布式锁使用及说明

《Redis分布式锁使用及说明》本文总结了Redis和Zookeeper在高可用性和高一致性场景下的应用,并详细介绍了Redis的分布式锁实现方式,包括使用Lua脚本和续期机制,最后,提到了RedLo... 目录Redis分布式锁加锁方式怎么会解错锁？举个小案例吧解锁方式续期总结Redis分布式锁如果追求
阅读更多...
Ubuntu 怎么启用 Universe 和 Multiverse 软件源?

Ubuntu 怎么启用 Universe 和 Multiverse 软件源?

《Ubuntu怎么启用Universe和Multiverse软件源?》在Ubuntu中，软件源是用于获取和安装软件的服务器，通过设置和管理软件源，您可以确保系统能够从可靠的来源获取最新的软件... Ubuntu 是一款广受认可且声誉良好的开源操作系统，允许用户通过其庞大的软件包来定制和增强计算体验。这些软件
阅读更多...
软件设计师备考——计算机系统

软件设计师备考——计算机系统

学习内容源自「软件设计师」 上午题 #1 计算机系统_哔哩哔哩_bilibili 目录 1.1.1 计算机系统硬件基本组成 1.1.2 中央处理单元 1.CPU 的功能 1)运算器 2)控制器 RISC && CISC 流水线控制 存储器  Cache 中断 输入输出IO控制方式 程序查询方式 中断驱动方式 直接存储器方式（DMA)  ​编辑 总线 ​编辑
阅读更多...
【STM32】SPI通信-软件与硬件读写SPI

【STM32】SPI通信-软件与硬件读写SPI

SPI通信-软件与硬件读写SPI 软件SPI一、SPI通信协议1、SPI通信2、硬件电路3、移位示意图4、SPI时序基本单元(1)开始通信和结束通信(2)模式0---用的最多(3)模式1(4)模式2(5)模式3 5、SPI时序(1)写使能(2)指定地址写(3)指定地址读 二、W25Q64模块介绍１、W25Q64简介２、硬件电路３、W25Q64框图４、Flash操作注意事项软件SPI读写W2
阅读更多...
免费也能高质量！2024年免费录屏软件深度对比评测

免费也能高质量！2024年免费录屏软件深度对比评测

我公司因为客户覆盖面广的原因经常会开远程会议，有时候说的内容比较广需要引用多份的数据，我记录起来有一定难度，所以一般都用录屏工具来记录会议内容。这次我们来一起探索有什么免费录屏工具可以提高我们的工作效率吧。 1.福晰录屏大师 链接直达：https://www.foxitsoftware.cn/REC/  录屏软件录屏功能就是本职，这款录屏工具在录屏模式上提供了多种选项，可以选择屏幕录制、窗口
阅读更多...
集中式版本控制与分布式版本控制——Git 学习笔记01

集中式版本控制与分布式版本控制——Git 学习笔记01

什么是版本控制 如果你用 Microsoft Word 写过东西，那你八成会有这样的经历： 想删除一段文字，又怕将来这段文字有用，怎么办呢？有一个办法，先把当前文件“另存为”一个文件，然后继续改，改到某个程度，再“另存为”一个文件。就这样改着、存着……最后你的 Word 文档变成了这样： 过了几天，你想找回被删除的文字，但是已经记不清保存在哪个文件了，只能挨个去找。真麻烦，眼睛都花了。看
阅读更多...
HomeBank：开源免费的个人财务管理软件

HomeBank：开源免费的个人财务管理软件

在个人财务管理领域，找到一个既免费又开源的解决方案并非易事。HomeBank&nbsp;正是这样一个项目，它不仅提供了强大的功能，还拥有一个活跃的社区，不断推动其发展和完善。 开源免费：HomeBank 是一个完全开源的项目，用户可以自由地使用、修改和分发。用户友好的界面：提供直观的图形用户界面，使得非技术用户也能轻松上手。数据导入支持：支持从 Quicken、Microsoft Money
阅读更多...
开源分布式数据库中间件

开源分布式数据库中间件

转自：https://www.csdn.net/article/2015-07-16/2825228 MyCat：开源分布式数据库中间件 为什么需要MyCat？ 虽然云计算时代，传统数据库存在着先天性的弊端，但是NoSQL数据库又无法将其替代。如果传统数据易于扩展，可切分，就可以避免单机（单库）的性能缺陷。 MyCat的目标就是：低成本地将现有的单机数据库和应用平滑迁移到“云”端
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2