Commit feb95858 authored by 赵威's avatar 赵威

save diary

parent d849ca0b
......@@ -45,6 +45,7 @@ object Main extends LazyLogging {
"172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
)
// TODO remove test
val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
"gm-portrait-update-device-test",
new UserInfoDeserializationSchema,
......
......@@ -14,9 +14,10 @@ object Redis {
implicit val ec: scala.concurrent.ExecutionContext = scala.concurrent.ExecutionContext.global
val pRc2 = new RedisClient("172.16.40.173", 6379, 0, Some("ReDis!GmTx*0aN9"), 5000)
val pRc4 = new RedisClient("172.16.50.145", 6379, 0, Some("XfkMCCdWDIU%ls$h"), 5000)
val pRc4 =
new RedisClientPool("172.16.50.145", 6379, database = 0, secret = Some("XfkMCCdWDIU%ls$h"), timeout = 500)
val pRc4Pool =
new RedisClientPool("172.16.50.145", 6379, database = 0, secret = Some("XfkMCCdWDIU%ls$h"), timeout = 5000)
def getRead(deviceId: String, contentType: String): List[Long] = {
val key = s"doris:user_portrait:tag3:read_v2:device_id:${deviceId}:content_type:${contentType}"
......@@ -36,7 +37,7 @@ object Redis {
contentEitherFuture match {
case Left(e) => e.printStackTrace()
case Right(contentFuture) =>
pRc4.withClient { client =>
pRc4Pool.withClient { client =>
contentFuture.foreach { seq =>
val ids = seq.map(c => c.id)
if (ids.size > 0) {
......@@ -75,14 +76,12 @@ object Redis {
contentEitherFuture match {
case Left(e) => e.printStackTrace()
case Right(contentFuture) =>
pRc4.withClient { client =>
contentFuture.foreach { seq =>
val idPairs = seq.map(c => (c.id, c.serviceId.getOrElse("-1")))
val res = write(idPairs)
if (idPairs.size > 0) {
client.del(key)
client.set(key, res)
}
contentFuture.foreach { seq =>
val idPairs = seq.map(c => (c.id, c.serviceId.getOrElse("-1")))
val res = write(idPairs)
if (idPairs.size > 0) {
pRc4.del(key)
pRc4.set(key, res)
}
}
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment